Compare commits
21 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| eb21a02b6d | |||
| 498ce4e287 | |||
| 7081a8e900 | |||
| da4b86a159 | |||
| 81d864f4bc | |||
| c9795a6c4d | |||
| f5dc55f1d1 | |||
| fd92df486c | |||
| fc7498fef0 | |||
| 51dcca592d | |||
| 27c1e18e98 | |||
| 73502db9f4 | |||
| 4f85ef5209 | |||
| def18f28fa | |||
| 8fc27f4d69 | |||
| 6137657704 | |||
| 704a8ab7de | |||
| 7f59b7fd35 | |||
| 660fc20124 | |||
| 07457ad556 | |||
| e9c4f23ae2 |
@@ -61,6 +61,7 @@ import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
@@ -89,6 +90,19 @@ API = f"https://{GITEA_HOST}/api/v1" if GITEA_HOST else ""
|
||||
# match by exact title without parsing.
|
||||
TITLE_PREFIX = "[main-red]"
|
||||
|
||||
# Settling window (seconds) between initial red detection and the
|
||||
# pre-file recheck. The recheck filters out the two largest false-
|
||||
# positive classes seen in mc#1597..1630 (task #394, 2026-05-21):
|
||||
# 1. HEAD moved on (a new commit landed mid-tick) — the prior red SHA
|
||||
# is no longer authoritative; let the next cron tick re-evaluate.
|
||||
# 2. Combined status recovered on the SAME SHA (transient
|
||||
# cancel-cascade rolled forward to success on retry).
|
||||
# 90s is well below the hourly cron cadence; a real failure that
|
||||
# persists past it is the one we want surfaced.
|
||||
# Override with WATCHDOG_RECHECK_DELAY_SECS for tests / local probes
|
||||
# (the test suite stubs time.sleep to a no-op).
|
||||
RECHECK_DELAY_SECS = int(_env("WATCHDOG_RECHECK_DELAY_SECS", default="90"))
|
||||
|
||||
|
||||
def _require_runtime_env() -> None:
|
||||
"""Enforce env contract — called from `main()` only.
|
||||
@@ -172,6 +186,49 @@ def api(
|
||||
return status, {"_raw": raw.decode("utf-8", errors="replace")}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# action_run.status resolver — extensibility hook for task #394.
|
||||
# --------------------------------------------------------------------------
|
||||
def _resolve_action_run_status(target_url: str) -> int | None:
|
||||
"""Resolve the underlying Gitea `action_run.status` integer for the
|
||||
run referenced by `target_url`, returning None if the resolver
|
||||
cannot reach an authoritative source from the runner.
|
||||
|
||||
Canonical Gitea 1.22.6 enum (per `models/actions/status.go` +
|
||||
`reference_gitea_action_status_enum_corrected_2026_05_19`):
|
||||
1=Success, 2=Failure, 3=Cancelled, 4=Skipped,
|
||||
5=Waiting, 6=Running, 7=Blocked
|
||||
Only `status == 2` is a real defect; status=3 is cancel-cascade and
|
||||
status=1 is an emission artifact (Gitea wrote a 'failure' commit_status
|
||||
row for a run that actually succeeded — observed empirically on
|
||||
`publish-canvas-image` jobs at SHAs in mc#1597..1630).
|
||||
|
||||
CURRENT STATE (2026-05-20, verified): Gitea 1.22.6 exposes NO REST
|
||||
endpoint for `action_run.status`. Probed:
|
||||
/api/v1/repos/{o}/{r}/actions/runs/{id} → HTTP 404
|
||||
/api/v1/repos/{o}/{r}/actions/jobs/{id} → HTTP 404
|
||||
/api/v1/repos/{o}/{r}/actions/tasks/{id} → HTTP 404
|
||||
/swagger.v1.json paths containing 'actions' → secrets+variables+runners only
|
||||
The SPA backend (`/{repo}/actions/runs/{id}/jobs/{idx}` POST) requires
|
||||
a session CSRF token, unreachable from a runner. The only authoritative
|
||||
source today is direct DB access (`mol_action_status` on op-host,
|
||||
`docker exec molecule-postgres-1 psql ...`), which the runner cannot
|
||||
reach.
|
||||
|
||||
Therefore: this hook returns None on every call. Callers MUST fall
|
||||
back to the description-string filter (existing) plus the HEAD
|
||||
recheck (this PR). When a future Gitea release (>=1.23 expected) or
|
||||
an op-host proxy exposes the endpoint, replace the body of this
|
||||
function with an `api(...)` call — the caller contract is stable.
|
||||
|
||||
See also:
|
||||
- `reference_chronic_red_sweep_cancelled_vs_failed_filter`
|
||||
- `feedback_gitea_status_enum_use_helper_not_raw_int`
|
||||
"""
|
||||
_ = target_url # noqa: F841 — intentional placeholder
|
||||
return None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Gitea reads
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -614,6 +671,56 @@ def run_once(*, dry_run: bool = False) -> int:
|
||||
}
|
||||
|
||||
if red:
|
||||
# HEAD recheck (task #394 — guards mc#1597..1630 false-positive
|
||||
# cluster). After the initial detection, wait RECHECK_DELAY_SECS
|
||||
# (default 90s; tests stub time.sleep) and re-evaluate:
|
||||
#
|
||||
# 1. Re-fetch HEAD SHA. If HEAD moved, a new commit landed
|
||||
# mid-tick — the prior red SHA is no longer authoritative
|
||||
# and the next cron run will re-evaluate against the new
|
||||
# HEAD. Skip-file.
|
||||
#
|
||||
# 2. If HEAD unchanged, re-fetch the combined status. If it
|
||||
# recovered (combined state no longer in {failure,error}
|
||||
# after the cancel-cascade filter), a transient retry
|
||||
# rolled the run forward. Skip-file.
|
||||
#
|
||||
# Both paths emit a Loki event distinguishable from the real
|
||||
# `main_red_detected` so obs queries can track filter activity.
|
||||
# The settling window is well below the hourly cron cadence —
|
||||
# genuine failures persist past it and are surfaced normally.
|
||||
time.sleep(RECHECK_DELAY_SECS)
|
||||
|
||||
recheck_sha = get_head_sha(WATCH_BRANCH)
|
||||
if recheck_sha != sha:
|
||||
emit_loki_event("main_red_skipped_head_drift", sha, [])
|
||||
print(
|
||||
f"::notice::skip-file (HEAD moved): initial red at "
|
||||
f"{sha[:10]} but HEAD is now {recheck_sha[:10]} on "
|
||||
f"{WATCH_BRANCH}; next cron tick will re-evaluate."
|
||||
)
|
||||
return 0
|
||||
|
||||
recheck_status = get_combined_status(sha)
|
||||
recheck_red, recheck_failed = is_red(recheck_status)
|
||||
if not recheck_red:
|
||||
emit_loki_event("main_red_skipped_recovered", sha, [])
|
||||
print(
|
||||
f"::notice::skip-file (recovered after settling): "
|
||||
f"combined state at {sha[:10]} flipped to "
|
||||
f"{recheck_status.get('state')!r} on recheck; "
|
||||
f"initial red was a transient cancel-cascade."
|
||||
)
|
||||
return 0
|
||||
|
||||
# Still red after settling — file/update. Use the recheck data
|
||||
# as authoritative so the issue body reflects the latest state.
|
||||
failed = recheck_failed
|
||||
debug["recheck_combined_state"] = recheck_status.get("state")
|
||||
debug["recheck_failed_contexts"] = [
|
||||
s.get("context") for s in failed
|
||||
]
|
||||
|
||||
failed_ctxs = [s.get("context") for s in failed if s.get("context")]
|
||||
emit_loki_event("main_red_detected", sha, failed_ctxs)
|
||||
print(f"::warning::main is RED at {sha[:10]} on {WATCH_BRANCH}: "
|
||||
|
||||
@@ -47,7 +47,9 @@ What this script does, per `.gitea/workflows/status-reaper.yml` invocation:
|
||||
Parse context as `<workflow_name> / <job_name> (push)`.
|
||||
Look up workflow_name in the trigger map:
|
||||
- missing → log ::notice:: and skip (conservative).
|
||||
- has_push_trigger=True → preserve (real defect signal).
|
||||
- has_push_trigger=True and description == "Has been cancelled"
|
||||
→ compensate cancelled/superseded push noise.
|
||||
- has_push_trigger=True otherwise → preserve (real defect signal).
|
||||
- has_push_trigger=False → POST a compensating
|
||||
`state=success` status to /statuses/{sha} with the same
|
||||
context (Gitea de-dups by context) and a description
|
||||
@@ -141,6 +143,11 @@ PR_SHADOW_COMPENSATION_DESCRIPTION = (
|
||||
"shadowed by successful push status on same SHA; see "
|
||||
".gitea/scripts/status-reaper.py)"
|
||||
)
|
||||
CANCELLED_PUSH_COMPENSATION_DESCRIPTION = (
|
||||
"Compensated by status-reaper (push run was cancelled/superseded; "
|
||||
"Gitea 1.22.6 reports cancelled runs as failure statuses)"
|
||||
)
|
||||
CANCELLED_DESCRIPTION = "Has been cancelled"
|
||||
|
||||
# Context suffix the reaper acts on. Gitea hardcodes this for ALL
|
||||
# default-branch workflow runs.
|
||||
@@ -476,7 +483,7 @@ def reap(
|
||||
{compensated, preserved_real_push, preserved_unknown,
|
||||
preserved_non_failure, preserved_non_push_suffix,
|
||||
preserved_unparseable, compensated_pr_shadowed_by_push_success,
|
||||
preserved_pr_without_push_success,
|
||||
preserved_pr_without_push_success, compensated_cancelled_push,
|
||||
compensated_contexts: [<context>, ...]}
|
||||
|
||||
`compensated_contexts` is rev2-added so `reap_branch` can build
|
||||
@@ -490,6 +497,7 @@ def reap(
|
||||
"preserved_non_push_suffix": 0,
|
||||
"preserved_unparseable": 0,
|
||||
"compensated_pr_shadowed_by_push_success": 0,
|
||||
"compensated_cancelled_push": 0,
|
||||
"preserved_pr_without_push_success": 0,
|
||||
"compensated_contexts": [],
|
||||
}
|
||||
@@ -567,8 +575,27 @@ def reap(
|
||||
counters["preserved_unknown"] += 1
|
||||
continue
|
||||
|
||||
if (s.get("description") or "").strip() == CANCELLED_DESCRIPTION:
|
||||
# Gitea 1.22.6 maps cancelled action runs to failure commit
|
||||
# statuses. During merge bursts, older push runs can be
|
||||
# superseded and cancelled even though a newer run for the
|
||||
# same branch is the real signal. Compensate only the exact
|
||||
# Gitea cancellation description; real push failures remain red.
|
||||
post_compensating_status(
|
||||
sha,
|
||||
context,
|
||||
s.get("target_url"),
|
||||
description=CANCELLED_PUSH_COMPENSATION_DESCRIPTION,
|
||||
dry_run=dry_run,
|
||||
)
|
||||
counters["compensated"] += 1
|
||||
counters["compensated_cancelled_push"] += 1
|
||||
counters["compensated_contexts"].append(context)
|
||||
continue
|
||||
|
||||
if workflow_trigger_map[workflow_name]:
|
||||
# Real push trigger → real defect signal. Preserve.
|
||||
# Real push trigger with a non-cancelled failure description
|
||||
# remains a defect signal. Preserve.
|
||||
counters["preserved_real_push"] += 1
|
||||
continue
|
||||
|
||||
@@ -674,6 +701,7 @@ def reap_branch(
|
||||
"preserved_non_push_suffix": 0,
|
||||
"preserved_unparseable": 0,
|
||||
"compensated_pr_shadowed_by_push_success": 0,
|
||||
"compensated_cancelled_push": 0,
|
||||
"preserved_pr_without_push_success": 0,
|
||||
"compensated_per_sha": {},
|
||||
"skipped": True,
|
||||
@@ -689,6 +717,7 @@ def reap_branch(
|
||||
"preserved_non_push_suffix": 0,
|
||||
"preserved_unparseable": 0,
|
||||
"compensated_pr_shadowed_by_push_success": 0,
|
||||
"compensated_cancelled_push": 0,
|
||||
"preserved_pr_without_push_success": 0,
|
||||
"compensated_per_sha": {},
|
||||
}
|
||||
@@ -728,6 +757,7 @@ def reap_branch(
|
||||
"preserved_non_push_suffix",
|
||||
"preserved_unparseable",
|
||||
"compensated_pr_shadowed_by_push_success",
|
||||
"compensated_cancelled_push",
|
||||
"preserved_pr_without_push_success",
|
||||
):
|
||||
aggregate[key] += per_sha[key]
|
||||
|
||||
@@ -44,6 +44,8 @@ name: E2E Peer Visibility (literal MCP list_peers)
|
||||
# - No cross-repo `uses:` (feedback_gitea_cross_repo_uses_blocked). The
|
||||
# actions/checkout SHA is the one e2e-staging-canvas.yml already uses
|
||||
# successfully (a mirrored SHA — see #1277/PR#1292 root-cause).
|
||||
# - 2026-05-21 retrigger: verify fresh platform-tenant image after the
|
||||
# publish Buildx DOCKER_CONFIG fix restored staging-latest image updates.
|
||||
# - Per-SHA concurrency, not global (feedback_concurrency_group_per_sha).
|
||||
# - Workflow-level GITHUB_SERVER_URL pinned
|
||||
# (feedback_act_runner_github_server_url).
|
||||
|
||||
@@ -25,8 +25,11 @@ name: publish-workspace-server-image
|
||||
# staging-<sha>. Set repo variable or secret PROD_AUTO_DEPLOY_DISABLED=true
|
||||
# to stop production rollout while keeping image publishing enabled.
|
||||
#
|
||||
# ECR target: 153263036946.dkr.ecr.us-east-2.amazonaws.com/molecule-ai/*
|
||||
# Primary ECR target: 153263036946.dkr.ecr.us-east-2.amazonaws.com/molecule-ai/*
|
||||
# Optional staging tenant mirror target:
|
||||
# 004947743811.dkr.ecr.us-east-2.amazonaws.com/molecule-ai/platform-tenant
|
||||
# Required secrets: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AUTO_SYNC_TOKEN
|
||||
# Optional secrets: AWS_STAGING_ECR_ACCESS_KEY_ID, AWS_STAGING_ECR_SECRET_ACCESS_KEY
|
||||
#
|
||||
# mc#711: Docker daemon not accessible on ubuntu-latest runner (molecule-canonical-1
|
||||
# shows client-only in `docker info` — daemon not running). DinD mount is present but
|
||||
@@ -65,6 +68,7 @@ env:
|
||||
# use below in this repo's staging-verify.yml.
|
||||
IMAGE_NAME: ${{ vars.ECR_REGISTRY || '153263036946.dkr.ecr.us-east-2.amazonaws.com' }}/molecule-ai/platform
|
||||
TENANT_IMAGE_NAME: ${{ vars.ECR_REGISTRY || '153263036946.dkr.ecr.us-east-2.amazonaws.com' }}/molecule-ai/platform-tenant
|
||||
STAGING_TENANT_IMAGE_NAME: ${{ vars.STAGING_ECR_REGISTRY || '004947743811.dkr.ecr.us-east-2.amazonaws.com' }}/molecule-ai/platform-tenant
|
||||
|
||||
jobs:
|
||||
build-and-push:
|
||||
@@ -135,6 +139,18 @@ jobs:
|
||||
run: |
|
||||
echo "sha=${GITHUB_SHA::7}" >> "$GITHUB_OUTPUT"
|
||||
|
||||
# Keep Buildx state inside the job temp dir. The publish runner's
|
||||
# inherited DOCKER_CONFIG can point at a host-owned ECR config path
|
||||
# (/home/hongming/.docker-ecr), which caused setup-buildx-action to
|
||||
# fail before image build with EACCES creating buildx/certs.
|
||||
- name: Prepare writable Docker config
|
||||
run: |
|
||||
set -euo pipefail
|
||||
export DOCKER_CONFIG="$RUNNER_TEMP/docker-config"
|
||||
mkdir -p "$DOCKER_CONFIG/buildx/certs"
|
||||
echo "DOCKER_CONFIG=$DOCKER_CONFIG" >> "$GITHUB_ENV"
|
||||
docker buildx version
|
||||
|
||||
# Build + push platform image (inline ECR auth — mirrors the operator-host
|
||||
# approach; credentials come from GITHUB_SECRET_AWS_ACCESS_KEY_ID /
|
||||
# GITHUB_SECRET_AWS_SECRET_ACCESS_KEY in Gitea Actions).
|
||||
@@ -170,21 +186,46 @@ jobs:
|
||||
--push .
|
||||
|
||||
# Build + push tenant image (Go platform + Next.js canvas in one image).
|
||||
# When staging ECR publisher credentials are configured, push the same
|
||||
# build to the staging account too so fresh staging/E2E tenants can pull
|
||||
# without cross-account ECR permissions.
|
||||
- name: Build & push tenant image to ECR (staging-<sha> + staging-latest)
|
||||
env:
|
||||
TENANT_IMAGE_NAME: ${{ env.TENANT_IMAGE_NAME }}
|
||||
STAGING_TENANT_IMAGE_NAME: ${{ env.STAGING_TENANT_IMAGE_NAME }}
|
||||
TAG_SHA: staging-${{ steps.tags.outputs.sha }}
|
||||
TAG_LATEST: staging-latest
|
||||
GIT_SHA: ${{ github.sha }}
|
||||
REPO: ${{ github.repository }}
|
||||
AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
|
||||
AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
|
||||
AWS_STAGING_ECR_ACCESS_KEY_ID: ${{ secrets.AWS_STAGING_ECR_ACCESS_KEY_ID }}
|
||||
AWS_STAGING_ECR_SECRET_ACCESS_KEY: ${{ secrets.AWS_STAGING_ECR_SECRET_ACCESS_KEY }}
|
||||
AWS_DEFAULT_REGION: us-east-2
|
||||
run: |
|
||||
set -euo pipefail
|
||||
ECR_REGISTRY="${TENANT_IMAGE_NAME%%/*}"
|
||||
aws ecr get-login-password --region us-east-2 | \
|
||||
docker login --username AWS --password-stdin "${ECR_REGISTRY}"
|
||||
|
||||
build_tags=(
|
||||
--tag "${TENANT_IMAGE_NAME}:${TAG_SHA}"
|
||||
--tag "${TENANT_IMAGE_NAME}:${TAG_LATEST}"
|
||||
)
|
||||
if [ -n "${AWS_STAGING_ECR_ACCESS_KEY_ID:-}" ] && [ -n "${AWS_STAGING_ECR_SECRET_ACCESS_KEY:-}" ]; then
|
||||
STAGING_ECR_REGISTRY="${STAGING_TENANT_IMAGE_NAME%%/*}"
|
||||
AWS_ACCESS_KEY_ID="${AWS_STAGING_ECR_ACCESS_KEY_ID}" \
|
||||
AWS_SECRET_ACCESS_KEY="${AWS_STAGING_ECR_SECRET_ACCESS_KEY}" \
|
||||
aws ecr get-login-password --region us-east-2 | \
|
||||
docker login --username AWS --password-stdin "${STAGING_ECR_REGISTRY}"
|
||||
build_tags+=(
|
||||
--tag "${STAGING_TENANT_IMAGE_NAME}:${TAG_SHA}"
|
||||
--tag "${STAGING_TENANT_IMAGE_NAME}:${TAG_LATEST}"
|
||||
)
|
||||
else
|
||||
echo "::notice::Skipping staging ECR tenant push; AWS_STAGING_ECR_ACCESS_KEY_ID/AWS_STAGING_ECR_SECRET_ACCESS_KEY are not configured."
|
||||
fi
|
||||
|
||||
docker buildx build \
|
||||
--file ./workspace-server/Dockerfile.tenant \
|
||||
--build-arg NEXT_PUBLIC_PLATFORM_URL= \
|
||||
@@ -193,8 +234,7 @@ jobs:
|
||||
--label "org.opencontainers.image.revision=${GIT_SHA}" \
|
||||
--label "org.opencontainers.image.created=$(date -u +%Y-%m-%dT%H:%M:%SZ)" \
|
||||
--label "molecule.workflow.run_id=${GITHUB_RUN_ID}" \
|
||||
--tag "${TENANT_IMAGE_NAME}:${TAG_SHA}" \
|
||||
--tag "${TENANT_IMAGE_NAME}:${TAG_LATEST}" \
|
||||
"${build_tags[@]}" \
|
||||
--push .
|
||||
|
||||
# bp-exempt: production deploy side-effect; merge is gated by CI / all-required and this job waits for push CI before acting.
|
||||
|
||||
+18
-1
@@ -127,7 +127,11 @@ cd workspace-server && go test -race ./...
|
||||
cd canvas && npm test
|
||||
|
||||
# Workspace runtime (Python)
|
||||
cd workspace && python -m pytest -v
|
||||
# Runtime code is SSOT in molecule-ai-workspace-runtime, not molecule-core/workspace.
|
||||
cd ../molecule-ai-workspace-runtime
|
||||
python -m venv .venv && source .venv/bin/activate
|
||||
pip install --index-url https://git.moleculesai.app/api/packages/molecule-ai/pypi/simple/ -e . pytest pytest-asyncio
|
||||
pytest -q
|
||||
|
||||
# E2E API tests (requires running platform)
|
||||
bash tests/e2e/test_api.sh
|
||||
@@ -159,6 +163,19 @@ and run CI manually.
|
||||
| review-check-tests | `review-check.sh` evaluator regression suite (13 scenarios) |
|
||||
| ops-scripts | Python unittest suite for `scripts/*.py` |
|
||||
|
||||
### Workspace runtime SSOT
|
||||
|
||||
Runtime code lives in
|
||||
[`molecule-ai-workspace-runtime`](https://git.moleculesai.app/molecule-ai/molecule-ai-workspace-runtime).
|
||||
Do not reintroduce `molecule-core/workspace/` or vendored `molecule_runtime/`
|
||||
copies in consumers. Core and templates consume the published runtime package
|
||||
from the Gitea package registry.
|
||||
|
||||
For local external MCP agents, multi-workspace config is
|
||||
`MOLECULE_WORKSPACES=[{"id":"...","token":"...","platform_url":"..."}]`.
|
||||
`platform_url` selects the tenant; `org_id` is not part of this config.
|
||||
Workspace IDs can differ across orgs.
|
||||
|
||||
## Local Testing
|
||||
|
||||
### review-check.sh
|
||||
|
||||
@@ -285,6 +285,39 @@ Canvas requests (no `X-Workspace-ID` header) and system callers
|
||||
|
||||
---
|
||||
|
||||
## Multiple Workspaces From One Local MCP Bridge
|
||||
|
||||
The standalone runtime package includes `molecule-mcp`, a local MCP bridge for
|
||||
external agents such as Claude Code, Codex, Hermes, and other tools that run
|
||||
outside the platform container fleet. One local bridge can serve multiple
|
||||
external workspaces by setting `MOLECULE_WORKSPACES`:
|
||||
|
||||
```json
|
||||
[
|
||||
{
|
||||
"id": "workspace-id-local-to-hongming-org",
|
||||
"token": "...",
|
||||
"platform_url": "https://hongming.moleculesai.app"
|
||||
},
|
||||
{
|
||||
"id": "different-workspace-id-local-to-agents-team-org",
|
||||
"token": "...",
|
||||
"platform_url": "https://agents-team.moleculesai.app"
|
||||
}
|
||||
]
|
||||
```
|
||||
|
||||
`platform_url` is the tenant routing key. The bridge registers, heartbeats,
|
||||
polls inboxes, and sends outbound A2A calls against the URL attached to the
|
||||
workspace that is doing the work.
|
||||
|
||||
Do not add `org_id` to this config. The tenant already comes from
|
||||
`platform_url`, and the bearer token is issued by that tenant. Workspace IDs
|
||||
also do not need to be shared across orgs; each tenant can return its own
|
||||
workspace ID and token for the same local agent process.
|
||||
|
||||
---
|
||||
|
||||
## Canvas Appearance
|
||||
|
||||
External workspaces appear on the canvas with a purple **REMOTE** badge
|
||||
|
||||
@@ -135,6 +135,33 @@ The `id` field is your workspace ID — remember it.
|
||||
|
||||
---
|
||||
|
||||
## Optional — one local MCP bridge, multiple tenants
|
||||
|
||||
If your local agent runtime uses `molecule-mcp`, one process can serve more
|
||||
than one external workspace:
|
||||
|
||||
```bash
|
||||
export MOLECULE_WORKSPACES='[
|
||||
{
|
||||
"id": "workspace-id-local-to-you-org",
|
||||
"token": "...",
|
||||
"platform_url": "https://you.moleculesai.app"
|
||||
},
|
||||
{
|
||||
"id": "different-workspace-id-local-to-team-org",
|
||||
"token": "...",
|
||||
"platform_url": "https://team.moleculesai.app"
|
||||
}
|
||||
]'
|
||||
molecule-mcp
|
||||
```
|
||||
|
||||
Use the workspace ID and token returned by each tenant. The IDs may differ
|
||||
across orgs. `org_id` is not required here because `platform_url` selects the
|
||||
tenant and the token is tenant-scoped.
|
||||
|
||||
---
|
||||
|
||||
## Step 4 — Chat with it
|
||||
|
||||
1. Open your Molecule canvas at `https://<TENANT>`
|
||||
|
||||
@@ -125,6 +125,33 @@ The agent appears on the canvas with a **purple REMOTE badge** within seconds. F
|
||||
|
||||
---
|
||||
|
||||
## Multi-Tenant Local MCP Bridge
|
||||
|
||||
For local MCP-driven agents, use the standalone runtime's `molecule-mcp`
|
||||
entrypoint. A single local bridge can serve multiple external workspaces by
|
||||
setting `MOLECULE_WORKSPACES`:
|
||||
|
||||
```json
|
||||
[
|
||||
{
|
||||
"id": "workspace-id-local-to-acme",
|
||||
"token": "...",
|
||||
"platform_url": "https://acme.moleculesai.app"
|
||||
},
|
||||
{
|
||||
"id": "different-workspace-id-local-to-ops",
|
||||
"token": "...",
|
||||
"platform_url": "https://ops.moleculesai.app"
|
||||
}
|
||||
]
|
||||
```
|
||||
|
||||
`platform_url` selects the tenant for registration, heartbeat, inbox polling,
|
||||
and outbound A2A routing. `org_id` is not required in this config, and the
|
||||
workspace IDs do not need to match across tenants.
|
||||
|
||||
---
|
||||
|
||||
## What Phase 30 Covers
|
||||
|
||||
| Phase | What shipped | Endpoint |
|
||||
|
||||
+1
-1
@@ -36,7 +36,7 @@ e2e_mint_test_token() {
|
||||
local admin_bearer="${MOLECULE_ADMIN_TOKEN:-${ADMIN_TOKEN:-}}"
|
||||
local admin_auth=()
|
||||
[ -n "$admin_bearer" ] && admin_auth=(-H "Authorization: Bearer $admin_bearer")
|
||||
body=$(curl -s -w "\n%{http_code}" "$BASE/admin/workspaces/$wid/test-token" "${admin_auth[@]}")
|
||||
body=$(curl -s -w "\n%{http_code}" "$BASE/admin/workspaces/$wid/test-token" ${admin_auth[@]+"${admin_auth[@]}"})
|
||||
local code
|
||||
code=$(printf '%s' "$body" | tail -n1)
|
||||
local json
|
||||
|
||||
@@ -71,7 +71,7 @@ pv_assert_runtime() {
|
||||
set +e
|
||||
resp=$(curl -sS -X POST "$base_url/workspaces/$wid/mcp" \
|
||||
-H "Authorization: Bearer $wtok" \
|
||||
"${org_header[@]}" \
|
||||
${org_header[@]+"${org_header[@]}"} \
|
||||
-H "Content-Type: application/json" \
|
||||
-d "$PV_RPC_BODY" \
|
||||
-o /tmp/pv_mcp_body.json -w "%{http_code}" 2>/dev/null)
|
||||
|
||||
@@ -24,7 +24,8 @@
|
||||
#
|
||||
# Only PROVISIONING differs from staging:
|
||||
# - staging: POST /cp/admin/orgs (cold EC2 tenant) + per-tenant admin
|
||||
# token + each workspace's auth_token from the POST /workspaces resp.
|
||||
# token + each workspace's MCP bearer from create response or an admin
|
||||
# token-mint fallback.
|
||||
# - local: POST /workspaces directly against the local stack
|
||||
# (BASE, default http://localhost:8080), MCP bearer minted via
|
||||
# GET /admin/workspaces/:id/test-token (e2e_mint_test_token —
|
||||
@@ -103,7 +104,7 @@ teardown() {
|
||||
log "[teardown] deleting ${#CREATED_WSIDS[@]} workspace(s) this run created (scoped)"
|
||||
for wid in ${CREATED_WSIDS[@]+"${CREATED_WSIDS[@]}"}; do
|
||||
[ -n "$wid" ] || continue
|
||||
curl -s -X DELETE "$BASE/workspaces/$wid?confirm=true" "${ADMIN_AUTH[@]}" >/dev/null 2>&1 || true
|
||||
curl -s -X DELETE "$BASE/workspaces/$wid?confirm=true" ${ADMIN_AUTH[@]+"${ADMIN_AUTH[@]}"} >/dev/null 2>&1 || true
|
||||
done
|
||||
exit $rc
|
||||
}
|
||||
@@ -112,7 +113,7 @@ trap teardown EXIT INT TERM
|
||||
# Pre-sweep workspaces a prior crashed run of THIS script left behind
|
||||
# (name prefix match only — never a blanket delete). The trap fires on
|
||||
# normal exit, but a kill -9 / SIGPIPE can bypass it.
|
||||
PRIOR=$(curl -s "$BASE/workspaces" "${ADMIN_AUTH[@]}" | python3 -c '
|
||||
PRIOR=$(curl -s "$BASE/workspaces" ${ADMIN_AUTH[@]+"${ADMIN_AUTH[@]}"} | python3 -c '
|
||||
import json, sys
|
||||
try:
|
||||
print(" ".join(w["id"] for w in json.load(sys.stdin) if w.get("name","").startswith("PV-Local-")))
|
||||
@@ -121,7 +122,7 @@ except Exception:
|
||||
' 2>/dev/null)
|
||||
for _wid in $PRIOR; do
|
||||
log "Pre-sweeping prior PV-Local workspace: $_wid"
|
||||
curl -s -X DELETE "$BASE/workspaces/$_wid?confirm=true" "${ADMIN_AUTH[@]}" >/dev/null 2>&1 || true
|
||||
curl -s -X DELETE "$BASE/workspaces/$_wid?confirm=true" ${ADMIN_AUTH[@]+"${ADMIN_AUTH[@]}"} >/dev/null 2>&1 || true
|
||||
done
|
||||
|
||||
# ─── Local-stack preflight ─────────────────────────────────────────────
|
||||
@@ -132,10 +133,10 @@ if ! curl -fsS "$BASE/health" -m 5 >/dev/null 2>&1; then
|
||||
fi
|
||||
# admin/test-token is the local MCP-bearer mint path; it 404s in
|
||||
# production. If it is off, this gate cannot drive the literal call.
|
||||
if ! curl -fsS "$BASE/admin/workspaces/preflight-probe/test-token" "${ADMIN_AUTH[@]}" -m 5 >/dev/null 2>&1; then
|
||||
if ! curl -fsS "$BASE/admin/workspaces/preflight-probe/test-token" ${ADMIN_AUTH[@]+"${ADMIN_AUTH[@]}"} -m 5 >/dev/null 2>&1; then
|
||||
# A 404 here is EITHER "no such ws" (fine — endpoint is enabled) OR the
|
||||
# endpoint is disabled (MOLECULE_ENV=production). Distinguish by body.
|
||||
PROBE=$(curl -s "$BASE/admin/workspaces/preflight-probe/test-token" "${ADMIN_AUTH[@]}" -m 5 2>/dev/null)
|
||||
PROBE=$(curl -s "$BASE/admin/workspaces/preflight-probe/test-token" ${ADMIN_AUTH[@]+"${ADMIN_AUTH[@]}"} -m 5 2>/dev/null)
|
||||
if echo "$PROBE" | grep -qi 'production\|disabled\|not found.*endpoint'; then
|
||||
echo "::error::GET /admin/workspaces/:id/test-token disabled (MOLECULE_ENV=production?). Cannot mint a local MCP bearer." >&2
|
||||
exit 1
|
||||
@@ -240,7 +241,7 @@ else
|
||||
fi
|
||||
log "1/5 provisioning parent ($PARENT_RUNTIME, mode=$PV_LOCAL_PROVISION_MODE) + one sibling per runtime under test..."
|
||||
|
||||
P_RESP=$(curl -s -X POST "$BASE/workspaces" "${ADMIN_AUTH[@]}" -H "Content-Type: application/json" \
|
||||
P_RESP=$(curl -s -X POST "$BASE/workspaces" ${ADMIN_AUTH[@]+"${ADMIN_AUTH[@]}"} -H "Content-Type: application/json" \
|
||||
-d "{\"name\":\"${NAME_PREFIX}-parent\",\"runtime\":\"$PARENT_RUNTIME\",\"tier\":3$PARENT_EXTRA,\"secrets\":$PARENT_SECRETS}")
|
||||
PARENT_ID=$(echo "$P_RESP" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("id",""))' 2>/dev/null)
|
||||
if [ -z "$PARENT_ID" ]; then
|
||||
@@ -290,7 +291,7 @@ for rt in $PV_RUNTIMES; do
|
||||
CREATE_RUNTIME="$rt"
|
||||
CREATE_EXTRA=""
|
||||
fi
|
||||
R=$(curl -s -X POST "$BASE/workspaces" "${ADMIN_AUTH[@]}" -H "Content-Type: application/json" \
|
||||
R=$(curl -s -X POST "$BASE/workspaces" ${ADMIN_AUTH[@]+"${ADMIN_AUTH[@]}"} -H "Content-Type: application/json" \
|
||||
-d "{\"name\":\"${NAME_PREFIX}-$rt\",\"runtime\":\"$CREATE_RUNTIME\",\"tier\":2,\"parent_id\":\"$PARENT_ID\"$CREATE_EXTRA,\"secrets\":$SEC}")
|
||||
WID=$(echo "$R" | python3 -c 'import json,sys;print(json.load(sys.stdin).get("id",""))' 2>/dev/null)
|
||||
if [ -z "$WID" ]; then
|
||||
|
||||
@@ -40,8 +40,10 @@
|
||||
# drives: POST /cp/admin/orgs (provision), GET
|
||||
# /cp/admin/orgs/:slug/admin-token (per-tenant token), DELETE
|
||||
# /cp/admin/tenants/:slug (teardown). The per-tenant admin token drives
|
||||
# tenant workspace creation; each workspace's OWN auth_token (returned by
|
||||
# POST /workspaces) drives its MCP call.
|
||||
# tenant workspace creation; each workspace's OWN auth_token drives its
|
||||
# MCP call. External-like runtimes may return the token in POST
|
||||
# /workspaces; managed container runtimes usually require the admin token
|
||||
# mint fallback below.
|
||||
#
|
||||
# Required env:
|
||||
# MOLECULE_ADMIN_TOKEN CP admin bearer — Railway staging CP_ADMIN_API_TOKEN
|
||||
@@ -104,6 +106,46 @@ tenant_call() {
|
||||
-H "Content-Type: application/json" "$@"
|
||||
}
|
||||
|
||||
tenant_call_capture() {
|
||||
local method="$1" path="$2" out="$3"; shift 3
|
||||
curl -sS -o "$out" -w "%{http_code}" -X "$method" "$TENANT_URL$path" \
|
||||
-H "Authorization: Bearer $TENANT_TOKEN" \
|
||||
-H "X-Molecule-Org-Id: $ORG_ID" \
|
||||
-H "Content-Type: application/json" "$@"
|
||||
}
|
||||
|
||||
redact_token_body() {
|
||||
python3 -c '
|
||||
import json, re, sys
|
||||
raw = sys.stdin.read()
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
except Exception:
|
||||
print(re.sub(r"(?i)([a-z0-9_]*token)=([^&\\s]+)", r"\1=<redacted>", raw)[:500])
|
||||
raise SystemExit(0)
|
||||
|
||||
def scrub(v):
|
||||
if isinstance(v, dict):
|
||||
return {k: ("<redacted>" if "token" in k.lower() else scrub(val)) for k, val in v.items()}
|
||||
if isinstance(v, list):
|
||||
return [scrub(x) for x in v]
|
||||
return v
|
||||
|
||||
print(json.dumps(scrub(data), separators=(",", ":"))[:500])
|
||||
'
|
||||
}
|
||||
|
||||
extract_auth_token() {
|
||||
python3 -c "
|
||||
import sys, json
|
||||
try:
|
||||
d = json.load(sys.stdin)
|
||||
except Exception:
|
||||
print(''); sys.exit(0)
|
||||
print(d.get('auth_token') or d.get('connection', {}).get('auth_token') or '')
|
||||
" 2>/dev/null
|
||||
}
|
||||
|
||||
# ─── Scoped teardown ───────────────────────────────────────────────────
|
||||
# Deletes ONLY the org this run created (DELETE /cp/admin/tenants/$SLUG
|
||||
# with the {"confirm":$SLUG} fat-finger guard). Never a cluster-wide
|
||||
@@ -218,25 +260,31 @@ for rt in $PV_RUNTIMES; do
|
||||
R=$(tenant_call POST /workspaces \
|
||||
-d "{\"name\":\"pv-$rt\",\"runtime\":\"$rt\",\"tier\":2,\"parent_id\":\"$PARENT_ID\",\"secrets\":$SECRETS_JSON}")
|
||||
WID=$(echo "$R" | python3 -c "import sys,json; print(json.load(sys.stdin).get('id',''))" 2>/dev/null)
|
||||
# auth_token is top-level for container runtimes; external-like nest it
|
||||
# under connection.auth_token (verified vs staging response shape).
|
||||
WTOK=$(echo "$R" | python3 -c "
|
||||
import sys, json
|
||||
try: d = json.load(sys.stdin)
|
||||
except Exception: print(''); sys.exit(0)
|
||||
print(d.get('auth_token') or d.get('connection', {}).get('auth_token') or '')
|
||||
" 2>/dev/null)
|
||||
# External-like runtimes may return connection.auth_token on create.
|
||||
# Managed container runtimes usually return only id/status here, then
|
||||
# receive their bearer through registry/bootstrap; for this literal MCP
|
||||
# driver we mint an admin test token below.
|
||||
WTOK=$(echo "$R" | extract_auth_token)
|
||||
[ -n "$WID" ] || fail "$rt workspace create failed: $(echo "$R" | head -c 300)"
|
||||
TOKEN_DIAG=""
|
||||
if [ -z "$WTOK" ]; then
|
||||
TTOK_RESP=$(tenant_call GET "/admin/workspaces/$WID/test-token" 2>/dev/null || true)
|
||||
WTOK=$(echo "$TTOK_RESP" | python3 -c "
|
||||
import sys, json
|
||||
try: d = json.load(sys.stdin)
|
||||
except Exception: print(''); sys.exit(0)
|
||||
print(d.get('auth_token') or '')
|
||||
" 2>/dev/null)
|
||||
TTOK_FILE=$(mktemp)
|
||||
TTOK_CODE=$(tenant_call_capture POST "/admin/workspaces/$WID/tokens" "$TTOK_FILE" 2>/dev/null || echo "curl_error")
|
||||
TTOK_RESP=$(cat "$TTOK_FILE" 2>/dev/null || true)
|
||||
WTOK=$(echo "$TTOK_RESP" | extract_auth_token)
|
||||
TOKEN_DIAG="POST /admin/workspaces/$WID/tokens -> HTTP $TTOK_CODE body: $(echo "$TTOK_RESP" | redact_token_body)"
|
||||
rm -f "$TTOK_FILE"
|
||||
fi
|
||||
[ -n "$WTOK" ] || fail "$rt workspace did not return or mint an auth_token — cannot drive its MCP call (resp: $(echo "$R" | head -c 300))"
|
||||
if [ -z "$WTOK" ]; then
|
||||
TTOK_FILE=$(mktemp)
|
||||
TTOK_CODE=$(tenant_call_capture GET "/admin/workspaces/$WID/test-token" "$TTOK_FILE" 2>/dev/null || echo "curl_error")
|
||||
TTOK_RESP=$(cat "$TTOK_FILE" 2>/dev/null || true)
|
||||
WTOK=$(echo "$TTOK_RESP" | extract_auth_token)
|
||||
TOKEN_DIAG="${TOKEN_DIAG}
|
||||
GET /admin/workspaces/$WID/test-token -> HTTP $TTOK_CODE body: $(echo "$TTOK_RESP" | redact_token_body)"
|
||||
rm -f "$TTOK_FILE"
|
||||
fi
|
||||
[ -n "$WTOK" ] || fail "$rt workspace did not return or mint an auth_token — cannot drive its MCP call (create_resp: $(echo "$R" | redact_token_body); token_fallbacks: $TOKEN_DIAG)"
|
||||
WS_IDS[$rt]="$WID"
|
||||
WS_TOKENS[$rt]="$WTOK"
|
||||
ALL_WS_IDS="$ALL_WS_IDS $WID"
|
||||
|
||||
@@ -56,6 +56,21 @@ SCRIPT_PATH = (
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _stub_time_sleep(monkeypatch):
|
||||
"""Autouse: stub time.sleep across every test.
|
||||
|
||||
The watchdog's RECHECK_DELAY_SECS (default 90s) is wired into
|
||||
run_once() via time.sleep(). Without this stub, integration-style
|
||||
tests that exercise run_once() would each block for 90s — a
|
||||
pre-fix `pytest -q` ran in ~0.1s; the unstubbed equivalent took
|
||||
>4 minutes (task #394 review evidence). Stubbing here keeps the
|
||||
suite fast and deterministic without requiring every red-path test
|
||||
to remember the patch.
|
||||
"""
|
||||
monkeypatch.setattr("time.sleep", lambda s: None)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def wd_module():
|
||||
"""Import the script as a module under a known env."""
|
||||
@@ -809,3 +824,214 @@ def test_require_runtime_env_exits_when_missing(wd_module, monkeypatch):
|
||||
with pytest.raises(SystemExit) as excinfo:
|
||||
wd_module._require_runtime_env()
|
||||
assert excinfo.value.code == 2
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Action-run status filter + HEAD-recheck (task #394, mc#1597..1630)
|
||||
#
|
||||
# The existing cancel-cascade filter matched description=='Has been
|
||||
# cancelled' EXACTLY, but a 7-day DB sweep on 2026-05-20 showed that
|
||||
# only 76/702 (~11%) of action_run.status=3 (Cancelled) entries carry
|
||||
# that string — 89% are written as 'Failing after Ns', indistinguishable
|
||||
# from real action_run.status=2 (Failure) at the commit_status layer.
|
||||
#
|
||||
# Gitea 1.22.6 has NO REST endpoint exposing action_run.status, so the
|
||||
# canonical filter (status=2 only) cannot run from a Gitea Actions
|
||||
# runner. The next-best signal is the HEAD-recheck: re-fetch HEAD SHA
|
||||
# (or its combined status) right before filing. If HEAD moved on or
|
||||
# combined state recovered, the prior "red" was a transient
|
||||
# cancel-cascade and we skip-file.
|
||||
#
|
||||
# References:
|
||||
# - reference_chronic_red_sweep_cancelled_vs_failed_filter
|
||||
# - feedback_gitea_status_enum_use_helper_not_raw_int
|
||||
# - reference_gitea_action_status_enum_corrected_2026_05_19
|
||||
# - triage evidence 2026-05-21 04:55 (6 cancellation + 1 emission
|
||||
# artifact across mc#1597,1605,1609,1613,1626,1627,1630)
|
||||
# --------------------------------------------------------------------------
|
||||
def test_head_recheck_skips_file_when_head_moved(wd_module, monkeypatch, capsys):
|
||||
"""When initial tick sees red at SHA_A but HEAD has since moved to
|
||||
SHA_B (next commit landed mid-tick), the watchdog must NOT file.
|
||||
Re-evaluation happens on the next cron tick against the new SHA.
|
||||
|
||||
REGRESSION CLASS: this guards mc#1597..#1630 — 7 false-positives
|
||||
filed in 24h because cancel-cascade fired commit_status=failure
|
||||
rows on SHAs that were already superseded by new merges."""
|
||||
SHA_A = SHA_RED
|
||||
SHA_B = SHA_GREEN
|
||||
failed_ctx = [
|
||||
{"context": "ci/test", "status": "failure",
|
||||
"target_url": "/r/runs/100/jobs/0",
|
||||
"description": "Failing after 12s"},
|
||||
]
|
||||
# First branches read returns SHA_A; the second (recheck) returns SHA_B
|
||||
# → watchdog detects HEAD drift and skip-files.
|
||||
branches_responses = iter([
|
||||
(200, _branches_response(SHA_A)),
|
||||
(200, _branches_response(SHA_B)),
|
||||
])
|
||||
|
||||
def fake_api(method, path, *, body=None, query=None, expect_json=True):
|
||||
if method == "GET" and path == "/repos/owner/repo/branches/main":
|
||||
return next(branches_responses)
|
||||
if method == "GET" and path == f"/repos/owner/repo/commits/{SHA_A}/status":
|
||||
return (200, _combined_status("failure", failed_ctx))
|
||||
if method == "POST" and path == "/repos/owner/repo/issues":
|
||||
raise AssertionError(
|
||||
"watchdog filed a phantom issue despite HEAD moving away "
|
||||
"from the red SHA (regression: mc#1597..1630)"
|
||||
)
|
||||
if method == "GET" and path == "/repos/owner/repo/issues":
|
||||
return (200, [])
|
||||
raise AssertionError(f"unexpected api call: {method} {path}")
|
||||
|
||||
# Settling delay is no-op'd by the _stub_time_sleep autouse fixture.
|
||||
monkeypatch.setattr(wd_module, "api", fake_api)
|
||||
wd_module.run_once(dry_run=False)
|
||||
captured = capsys.readouterr()
|
||||
assert "head drift" in captured.out.lower() or "head moved" in captured.out.lower(), (
|
||||
f"expected a notice about HEAD drift, got: {captured.out!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_head_recheck_skips_file_when_recheck_status_recovered(
|
||||
wd_module, monkeypatch, capsys,
|
||||
):
|
||||
"""When initial tick sees red at SHA, but the post-settling recheck
|
||||
on the SAME SHA shows combined status recovered (e.g. transient
|
||||
cancel-cascade rolled forward to success on retry), skip-file.
|
||||
|
||||
This catches the mid-flight cancel-cascade window — the second
|
||||
largest false-positive cluster in mc#1597..1630."""
|
||||
failed_ctx_initial = [
|
||||
{"context": "ci/test", "status": "failure",
|
||||
"target_url": "/r/runs/100/jobs/0",
|
||||
"description": "Failing after 12s"},
|
||||
]
|
||||
recovered_ctx = [
|
||||
{"context": "ci/test", "status": "success",
|
||||
"target_url": "/r/runs/100/jobs/0",
|
||||
"description": "Successful in 30s"},
|
||||
]
|
||||
# Same SHA across both branch reads; status flips from failure→success
|
||||
# between the two combined-status reads.
|
||||
status_responses = iter([
|
||||
(200, _combined_status("failure", failed_ctx_initial)),
|
||||
(200, _combined_status("success", recovered_ctx)),
|
||||
])
|
||||
|
||||
def fake_api(method, path, *, body=None, query=None, expect_json=True):
|
||||
if method == "GET" and path == "/repos/owner/repo/branches/main":
|
||||
return (200, _branches_response(SHA_RED))
|
||||
if method == "GET" and path == f"/repos/owner/repo/commits/{SHA_RED}/status":
|
||||
return next(status_responses)
|
||||
if method == "POST" and path == "/repos/owner/repo/issues":
|
||||
raise AssertionError(
|
||||
"watchdog filed a phantom issue despite combined status "
|
||||
"recovering on recheck (mid-flight cancel-cascade window)"
|
||||
)
|
||||
if method == "GET" and path == "/repos/owner/repo/issues":
|
||||
return (200, [])
|
||||
raise AssertionError(f"unexpected api call: {method} {path}")
|
||||
|
||||
monkeypatch.setattr(wd_module, "api", fake_api)
|
||||
wd_module.run_once(dry_run=False)
|
||||
captured = capsys.readouterr()
|
||||
assert "recovered" in captured.out.lower() or "settled" in captured.out.lower(), (
|
||||
f"expected a notice about post-settling recovery, got: {captured.out!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_head_recheck_files_when_still_red_after_settling(
|
||||
wd_module, monkeypatch,
|
||||
):
|
||||
"""When BOTH the initial detection AND the post-settling recheck
|
||||
show the same SHA still red, file the issue. This is the genuine-
|
||||
failure path the watchdog is designed to surface.
|
||||
|
||||
Locks the over-filter: a future change that always-skips after
|
||||
recheck would dismiss real failures."""
|
||||
failed_ctx = [
|
||||
{"context": "ci/test", "status": "failure",
|
||||
"target_url": "/r/runs/100/jobs/0",
|
||||
"description": "Failing after 12s"},
|
||||
]
|
||||
post_filed = {"value": False}
|
||||
|
||||
def fake_api(method, path, *, body=None, query=None, expect_json=True):
|
||||
if method == "GET" and path == "/repos/owner/repo/branches/main":
|
||||
return (200, _branches_response(SHA_RED))
|
||||
if method == "GET" and path == f"/repos/owner/repo/commits/{SHA_RED}/status":
|
||||
return (200, _combined_status("failure", failed_ctx))
|
||||
if method == "GET" and path == "/repos/owner/repo/issues":
|
||||
return (200, [])
|
||||
if method == "GET" and path == "/repos/owner/repo/labels":
|
||||
return (200, [{"id": 9, "name": "tier:high"}])
|
||||
if method == "POST" and path == "/repos/owner/repo/issues":
|
||||
post_filed["value"] = True
|
||||
return (201, {"number": 999})
|
||||
if method == "POST" and path == "/repos/owner/repo/issues/999/labels":
|
||||
return (200, [])
|
||||
raise AssertionError(f"unexpected api call: {method} {path}")
|
||||
|
||||
monkeypatch.setattr(wd_module, "api", fake_api)
|
||||
wd_module.run_once(dry_run=False)
|
||||
assert post_filed["value"], (
|
||||
"genuine-failure path was skip-filed — head-recheck over-filter "
|
||||
"regression (would suppress all real main-red alarms)"
|
||||
)
|
||||
|
||||
|
||||
def test_head_recheck_skips_when_initial_was_only_cancel_cascade(
|
||||
wd_module, monkeypatch,
|
||||
):
|
||||
"""Belt-and-braces: combined-status failure caused exclusively by
|
||||
description='Has been cancelled' entries should still be filtered
|
||||
by the EXISTING cancel-cascade filter — head-recheck must not
|
||||
accidentally bypass it. Regression guard for the existing mc#1564
|
||||
fix."""
|
||||
failed_ctx = [
|
||||
{"context": "ci/test", "status": "failure",
|
||||
"description": "Has been cancelled"},
|
||||
]
|
||||
|
||||
def fake_api(method, path, *, body=None, query=None, expect_json=True):
|
||||
if method == "GET" and path == "/repos/owner/repo/branches/main":
|
||||
return (200, _branches_response(SHA_RED))
|
||||
if method == "GET" and path == f"/repos/owner/repo/commits/{SHA_RED}/status":
|
||||
return (200, _combined_status("failure", failed_ctx))
|
||||
if method == "POST" and path == "/repos/owner/repo/issues":
|
||||
raise AssertionError(
|
||||
"cancel-cascade-only entry must be filtered before any "
|
||||
"head-recheck logic runs"
|
||||
)
|
||||
if method == "GET" and path == "/repos/owner/repo/issues":
|
||||
return (200, [])
|
||||
# No commit-status recheck should happen because is_red() returned False
|
||||
raise AssertionError(f"unexpected api call: {method} {path}")
|
||||
|
||||
monkeypatch.setattr(wd_module, "api", fake_api)
|
||||
wd_module.run_once(dry_run=False)
|
||||
# success: no AssertionError raised, no POST
|
||||
|
||||
|
||||
def test_resolve_action_run_status_returns_none_on_no_endpoint(wd_module):
|
||||
"""The action_run.status REST endpoint does NOT exist in Gitea
|
||||
1.22.6 (verified empirically 2026-05-20 — /api/v1/.../actions/runs/N
|
||||
returns HTTP 404 across all probe variants). The resolver must
|
||||
return None gracefully so callers fall back to the description-
|
||||
string + head-recheck heuristics.
|
||||
|
||||
This pins the extensibility hook: when a future Gitea release (or
|
||||
an op-host proxy) exposes the endpoint, the resolver implementation
|
||||
can be swapped in without touching the caller contract."""
|
||||
# The function exists and is callable
|
||||
assert hasattr(wd_module, "_resolve_action_run_status")
|
||||
# A typical target_url shape from real Gitea commit_status rows:
|
||||
target_url = "/molecule-ai/molecule-core/actions/runs/75020/jobs/0"
|
||||
# Return None when no endpoint available
|
||||
out = wd_module._resolve_action_run_status(target_url)
|
||||
assert out is None, (
|
||||
"resolver must return None when the action_run.status endpoint "
|
||||
"isn't reachable — callers depend on the None-fallback path"
|
||||
)
|
||||
|
||||
@@ -442,6 +442,46 @@ def test_reap_preserves_real_push(sr_module, monkeypatch):
|
||||
assert calls == [] # NO POST
|
||||
|
||||
|
||||
def test_reap_compensates_cancelled_real_push_status(sr_module, monkeypatch):
|
||||
"""Gitea 1.22.6 maps cancelled push runs to failure statuses.
|
||||
|
||||
A real push workflow with description exactly "Has been cancelled"
|
||||
is cancel-cascade noise, not a defect signal. Status-reaper should
|
||||
compensate it even though the workflow has a push trigger.
|
||||
"""
|
||||
calls = []
|
||||
|
||||
def fake_api(method, path, *, body=None, query=None, expect_json=True):
|
||||
calls.append((method, path, body))
|
||||
return (201, {})
|
||||
|
||||
monkeypatch.setattr(sr_module, "api", fake_api)
|
||||
|
||||
workflow_map = {"ci": True}
|
||||
combined = {
|
||||
"state": "failure",
|
||||
"statuses": [
|
||||
{
|
||||
"context": "ci / test (push)",
|
||||
"status": "failure",
|
||||
"description": "Has been cancelled",
|
||||
"target_url": "https://example.test/actions/runs/1",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
counters = sr_module.reap(workflow_map, combined, SHA, dry_run=False)
|
||||
|
||||
assert counters["compensated"] == 1
|
||||
assert counters["compensated_cancelled_push"] == 1
|
||||
assert counters["preserved_real_push"] == 0
|
||||
assert len(calls) == 1
|
||||
assert calls[0][0] == "POST"
|
||||
assert calls[0][1] == f"/repos/owner/repo/statuses/{SHA}"
|
||||
assert calls[0][2]["context"] == "ci / test (push)"
|
||||
assert calls[0][2]["state"] == "success"
|
||||
|
||||
|
||||
def test_reap_preserves_unknown_workflow(sr_module, monkeypatch, capsys):
|
||||
"""Workflow not in map → ::notice:: + skip (conservative)."""
|
||||
monkeypatch.setattr(
|
||||
|
||||
@@ -686,11 +686,22 @@ func (h *WorkspaceHandler) resolveAgentURL(ctx context.Context, workspaceID stri
|
||||
_ = db.CacheURL(ctx, workspaceID, agentURL)
|
||||
}
|
||||
|
||||
// When the platform runs inside Docker, 127.0.0.1:{host_port} is
|
||||
// unreachable (it's the platform container's own localhost, not the
|
||||
// Docker host). Rewrite to the container's Docker-bridge hostname.
|
||||
// When the platform runs inside Docker, a managed workspace's
|
||||
// 127.0.0.1:{host_port} URL points at the Docker host and must be
|
||||
// rewritten to the workspace container's Docker-bridge hostname.
|
||||
// External runtimes are not managed containers; their local test/runtime
|
||||
// URL is the target and must not be synthesized into ws-<id>:8000.
|
||||
if strings.HasPrefix(agentURL, "http://127.0.0.1:") && h.provisioner != nil && platformInDocker {
|
||||
agentURL = provisioner.InternalURL(workspaceID)
|
||||
var wsRuntime string
|
||||
if err := db.DB.QueryRowContext(ctx,
|
||||
`SELECT COALESCE(runtime, 'langgraph') FROM workspaces WHERE id = $1`,
|
||||
workspaceID,
|
||||
).Scan(&wsRuntime); err != nil {
|
||||
log.Printf("ProxyA2A: runtime lookup before Docker URL rewrite failed for %s: %v", workspaceID, err)
|
||||
}
|
||||
if !isExternalLikeRuntime(wsRuntime) {
|
||||
agentURL = provisioner.InternalURL(workspaceID)
|
||||
}
|
||||
}
|
||||
// SSRF defence: reject private/metadata URLs before making outbound call.
|
||||
if err := isSafeURL(agentURL); err != nil {
|
||||
|
||||
@@ -1511,6 +1511,35 @@ func TestResolveAgentURL_DockerRewrite(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveAgentURL_ExternalRuntimeLoopbackNotRewrittenInDocker(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
mr := setupTestRedis(t)
|
||||
allowLoopbackForTest(t)
|
||||
handler := NewWorkspaceHandler(newTestBroadcaster(), nil, "http://localhost:8080", t.TempDir())
|
||||
waitForHandlerAsyncBeforeDBCleanup(t, handler)
|
||||
handler.provisioner = &stubLocalProv{}
|
||||
|
||||
restore := setPlatformInDockerForTest(true)
|
||||
defer restore()
|
||||
|
||||
agentURL := "http://127.0.0.1:55555"
|
||||
mr.Set("ws:ws-external:url", agentURL)
|
||||
mock.ExpectQuery("SELECT COALESCE\\(runtime").
|
||||
WithArgs("ws-external").
|
||||
WillReturnRows(sqlmock.NewRows([]string{"runtime"}).AddRow("external"))
|
||||
|
||||
url, perr := handler.resolveAgentURL(context.Background(), "ws-external")
|
||||
if perr != nil {
|
||||
t.Fatalf("unexpected error: %+v", perr)
|
||||
}
|
||||
if url != agentURL {
|
||||
t.Errorf("external runtime loopback URL must not be rewritten; got %q want %q", url, agentURL)
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet sqlmock expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// --- dispatchA2A direct unit tests ---
|
||||
|
||||
func TestDispatchA2A_BuildRequestError(t *testing.T) {
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/wsauth"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// AdminWorkspaceTokenHandler lets tenant admins mint the first workspace
|
||||
// bearer for managed SaaS workspaces whose runtime receives its token later
|
||||
// through registry registration.
|
||||
type AdminWorkspaceTokenHandler struct{}
|
||||
|
||||
func NewAdminWorkspaceTokenHandler() *AdminWorkspaceTokenHandler {
|
||||
return &AdminWorkspaceTokenHandler{}
|
||||
}
|
||||
|
||||
// Create handles POST /admin/workspaces/:id/tokens. The route must be mounted
|
||||
// behind AdminAuth; the plaintext token is returned exactly once.
|
||||
func (h *AdminWorkspaceTokenHandler) Create(c *gin.Context) {
|
||||
workspaceID := c.Param("id")
|
||||
if !validWorkspaceID(workspaceID) {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid workspace id"})
|
||||
return
|
||||
}
|
||||
|
||||
var existing string
|
||||
err := db.DB.QueryRowContext(c.Request.Context(),
|
||||
`SELECT id FROM workspaces WHERE id = $1 AND status <> 'removed'`,
|
||||
workspaceID).Scan(&existing)
|
||||
if err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "workspace not found"})
|
||||
return
|
||||
}
|
||||
log.Printf("admin workspace tokens: workspace lookup failed for %s: %v", workspaceID, err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "workspace lookup failed"})
|
||||
return
|
||||
}
|
||||
|
||||
var count int
|
||||
if err := db.DB.QueryRowContext(c.Request.Context(),
|
||||
`SELECT COUNT(*) FROM workspace_auth_tokens WHERE workspace_id = $1 AND revoked_at IS NULL`,
|
||||
workspaceID).Scan(&count); err != nil {
|
||||
log.Printf("admin workspace tokens: count failed for %s: %v", workspaceID, err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to count tokens"})
|
||||
return
|
||||
}
|
||||
if count >= maxTokensPerWorkspace {
|
||||
c.JSON(http.StatusTooManyRequests, gin.H{"error": fmt.Sprintf("maximum %d active tokens per workspace", maxTokensPerWorkspace)})
|
||||
return
|
||||
}
|
||||
|
||||
token, err := wsauth.IssueToken(c.Request.Context(), db.DB, workspaceID)
|
||||
if err != nil {
|
||||
log.Printf("admin workspace tokens: issue failed for %s: %v", workspaceID, err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create token"})
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf("admin workspace tokens: issued token for workspace %s", workspaceID)
|
||||
c.JSON(http.StatusCreated, gin.H{
|
||||
"auth_token": token,
|
||||
"workspace_id": workspaceID,
|
||||
"message": "Save this token now — it cannot be retrieved again.",
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"testing"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func TestAdminWorkspaceTokenHandler_Create_HappyPath(t *testing.T) {
|
||||
mock, cleanup := withMockDB(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectQuery(`SELECT id FROM workspaces WHERE id = \$1 AND status <> 'removed'`).
|
||||
WithArgs(wsUUID1).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(wsUUID1))
|
||||
mock.ExpectQuery(`SELECT COUNT\(\*\) FROM workspace_auth_tokens`).
|
||||
WithArgs(wsUUID1).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
|
||||
mock.ExpectExec(`INSERT INTO workspace_auth_tokens`).
|
||||
WithArgs(wsUUID1, sqlmock.AnyArg(), sqlmock.AnyArg()).
|
||||
WillReturnResult(sqlmock.NewResult(1, 1))
|
||||
|
||||
w := makeReq(t, NewAdminWorkspaceTokenHandler().Create, "POST",
|
||||
"/admin/workspaces/"+wsUUID1+"/tokens", gin.Params{{Key: "id", Value: wsUUID1}})
|
||||
|
||||
if w.Code != http.StatusCreated {
|
||||
t.Fatalf("expected 201, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
var body struct {
|
||||
AuthToken string `json:"auth_token"`
|
||||
WorkspaceID string `json:"workspace_id"`
|
||||
}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if body.AuthToken == "" || body.WorkspaceID != wsUUID1 {
|
||||
t.Fatalf("unexpected body: %+v", body)
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminWorkspaceTokenHandler_Create_MissingWorkspace(t *testing.T) {
|
||||
mock, cleanup := withMockDB(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectQuery(`SELECT id FROM workspaces WHERE id = \$1 AND status <> 'removed'`).
|
||||
WithArgs(wsUUID1).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}))
|
||||
|
||||
w := makeReq(t, NewAdminWorkspaceTokenHandler().Create, "POST",
|
||||
"/admin/workspaces/"+wsUUID1+"/tokens", gin.Params{{Key: "id", Value: wsUUID1}})
|
||||
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Fatalf("expected 404, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminWorkspaceTokenHandler_Create_RateLimited(t *testing.T) {
|
||||
mock, cleanup := withMockDB(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectQuery(`SELECT id FROM workspaces WHERE id = \$1 AND status <> 'removed'`).
|
||||
WithArgs(wsUUID1).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(wsUUID1))
|
||||
mock.ExpectQuery(`SELECT COUNT\(\*\) FROM workspace_auth_tokens`).
|
||||
WithArgs(wsUUID1).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(maxTokensPerWorkspace))
|
||||
|
||||
w := makeReq(t, NewAdminWorkspaceTokenHandler().Create, "POST",
|
||||
"/admin/workspaces/"+wsUUID1+"/tokens", gin.Params{{Key: "id", Value: wsUUID1}})
|
||||
|
||||
if w.Code != http.StatusTooManyRequests {
|
||||
t.Fatalf("expected 429, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminWorkspaceTokenHandler_Create_IssueFails(t *testing.T) {
|
||||
mock, cleanup := withMockDB(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectQuery(`SELECT id FROM workspaces WHERE id = \$1 AND status <> 'removed'`).
|
||||
WithArgs(wsUUID1).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(wsUUID1))
|
||||
mock.ExpectQuery(`SELECT COUNT\(\*\) FROM workspace_auth_tokens`).
|
||||
WithArgs(wsUUID1).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
|
||||
mock.ExpectExec(`INSERT INTO workspace_auth_tokens`).
|
||||
WillReturnError(errors.New("disk full"))
|
||||
|
||||
w := makeReq(t, NewAdminWorkspaceTokenHandler().Create, "POST",
|
||||
"/admin/workspaces/"+wsUUID1+"/tokens", gin.Params{{Key: "id", Value: wsUUID1}})
|
||||
|
||||
if w.Code != http.StatusInternalServerError {
|
||||
t.Fatalf("expected 500, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
@@ -84,6 +84,7 @@ type mcpTool struct {
|
||||
type MCPHandler struct {
|
||||
database *sql.DB
|
||||
broadcaster *events.Broadcaster
|
||||
a2aProxy func(ctx context.Context, workspaceID string, body []byte, callerID string, logActivity bool) (int, []byte, error)
|
||||
|
||||
// memv2 is the v2 memory plugin wiring (RFC #2728). nil-safe:
|
||||
// every v2 tool calls memoryV2Available() first and returns a
|
||||
@@ -98,6 +99,14 @@ func NewMCPHandler(database *sql.DB, broadcaster *events.Broadcaster) *MCPHandle
|
||||
return &MCPHandler{database: database, broadcaster: broadcaster}
|
||||
}
|
||||
|
||||
func (h *MCPHandler) proxyA2ARequest(ctx context.Context, workspaceID string, body []byte, callerID string, logActivity bool) (int, []byte, error) {
|
||||
if h.a2aProxy != nil {
|
||||
return h.a2aProxy(ctx, workspaceID, body, callerID, logActivity)
|
||||
}
|
||||
wh := NewWorkspaceHandler(h.broadcaster, nil, "", "")
|
||||
return wh.ProxyA2ARequest(ctx, workspaceID, body, callerID, logActivity)
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Tool definitions (mirrors workspace/a2a_mcp_server.py TOOLS list)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -53,6 +53,15 @@ func mcpPost(t *testing.T, h *MCPHandler, workspaceID string, body interface{})
|
||||
return w
|
||||
}
|
||||
|
||||
func expectCanCommunicateSiblings(mock sqlmock.Sqlmock, callerID, targetID, parentID string) {
|
||||
mock.ExpectQuery(`SELECT id, parent_id FROM workspaces WHERE id = \$1`).
|
||||
WithArgs(callerID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id", "parent_id"}).AddRow(callerID, parentID))
|
||||
mock.ExpectQuery(`SELECT id, parent_id FROM workspaces WHERE id = \$1`).
|
||||
WithArgs(targetID).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id", "parent_id"}).AddRow(targetID, parentID))
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// initialize
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
@@ -178,6 +187,98 @@ func TestMCPHandler_ToolsList_ContainsExpectedTools(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMCPHandler_DelegateTask_RoutesThroughPlatformA2AProxy(t *testing.T) {
|
||||
h, mock := newMCPHandler(t)
|
||||
callerID := "11111111-1111-1111-1111-111111111111"
|
||||
targetID := "22222222-2222-2222-2222-222222222222"
|
||||
parentID := "33333333-3333-3333-3333-333333333333"
|
||||
|
||||
expectCanCommunicateSiblings(mock, callerID, targetID, parentID)
|
||||
mock.ExpectExec(`(?s)INSERT INTO activity_logs.*'delegation'.*'delegate'`).
|
||||
WithArgs(callerID, callerID, targetID, "Delegating to "+targetID, sqlmock.AnyArg()).
|
||||
WillReturnResult(sqlmock.NewResult(1, 1))
|
||||
mock.ExpectExec(`UPDATE activity_logs`).
|
||||
WithArgs("dispatched", "", callerID, sqlmock.AnyArg()).
|
||||
WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
var gotTarget, gotCaller string
|
||||
h.a2aProxy = func(ctx context.Context, workspaceID string, body []byte, callerID string, logActivity bool) (int, []byte, error) {
|
||||
gotTarget = workspaceID
|
||||
gotCaller = callerID
|
||||
if !logActivity {
|
||||
t.Fatal("delegate_task should log through platform A2A proxy")
|
||||
}
|
||||
if !strings.Contains(string(body), "do work") {
|
||||
t.Fatalf("A2A body missing task text: %s", string(body))
|
||||
}
|
||||
return 200, []byte(`{"result":{"message":{"parts":[{"text":"done"}]}}}`), nil
|
||||
}
|
||||
|
||||
out, err := h.toolDelegateTask(context.Background(), callerID, map[string]interface{}{
|
||||
"workspace_id": targetID,
|
||||
"task": "do work",
|
||||
}, mcpCallTimeout)
|
||||
if err != nil {
|
||||
t.Fatalf("delegate_task returned error: %v", err)
|
||||
}
|
||||
if out != "done" {
|
||||
t.Fatalf("delegate_task response = %q, want done", out)
|
||||
}
|
||||
if gotTarget != targetID || gotCaller != callerID {
|
||||
t.Fatalf("proxy called with target=%q caller=%q, want target=%q caller=%q", gotTarget, gotCaller, targetID, callerID)
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Fatalf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMCPHandler_DelegateTaskAsync_RoutesThroughPlatformA2AProxy(t *testing.T) {
|
||||
h, mock := newMCPHandler(t)
|
||||
callerID := "11111111-1111-1111-1111-111111111111"
|
||||
targetID := "22222222-2222-2222-2222-222222222222"
|
||||
parentID := "33333333-3333-3333-3333-333333333333"
|
||||
|
||||
expectCanCommunicateSiblings(mock, callerID, targetID, parentID)
|
||||
mock.ExpectExec(`(?s)INSERT INTO activity_logs.*'delegation'.*'delegate'`).
|
||||
WithArgs(callerID, callerID, targetID, "Delegating to "+targetID, sqlmock.AnyArg()).
|
||||
WillReturnResult(sqlmock.NewResult(1, 1))
|
||||
mock.ExpectExec(`UPDATE activity_logs`).
|
||||
WithArgs("dispatched", "", callerID, sqlmock.AnyArg()).
|
||||
WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
called := make(chan struct{}, 1)
|
||||
h.a2aProxy = func(ctx context.Context, workspaceID string, body []byte, proxyCallerID string, logActivity bool) (int, []byte, error) {
|
||||
if workspaceID != targetID || proxyCallerID != callerID {
|
||||
t.Fatalf("unexpected proxy route target=%q caller=%q", workspaceID, proxyCallerID)
|
||||
}
|
||||
if !strings.Contains(string(body), "async work") {
|
||||
t.Fatalf("A2A body missing task text: %s", string(body))
|
||||
}
|
||||
called <- struct{}{}
|
||||
return 200, []byte(`{"result":{"message":{"parts":[{"text":"accepted"}]}}}`), nil
|
||||
}
|
||||
|
||||
out, err := h.toolDelegateTaskAsync(context.Background(), callerID, map[string]interface{}{
|
||||
"workspace_id": targetID,
|
||||
"task": "async work",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("delegate_task_async returned error: %v", err)
|
||||
}
|
||||
if !strings.Contains(out, `"status":"dispatched"`) {
|
||||
t.Fatalf("delegate_task_async response = %s", out)
|
||||
}
|
||||
waitGlobalAsyncForTest()
|
||||
select {
|
||||
case <-called:
|
||||
default:
|
||||
t.Fatal("async delegate did not call platform A2A proxy")
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Fatalf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// notifications/initialized
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -7,24 +7,19 @@ package handlers
|
||||
// and A2A response parsing helpers.
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/provisioner"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/registry"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// insertMCPDelegationRow writes a delegation activity row so the canvas
|
||||
// Agent Comms tab can show the task text for MCP-initiated delegations.
|
||||
// Mirrors insertDelegationRow (delegation.go) for the MCP tool path.
|
||||
@@ -190,15 +185,6 @@ func (h *MCPHandler) toolDelegateTask(ctx context.Context, callerID string, args
|
||||
// Non-fatal: still make the A2A call even if activity log write fails.
|
||||
}
|
||||
|
||||
agentURL, err := mcpResolveURL(ctx, h.database, targetID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
// SSRF defence: reject private/metadata URLs before making outbound call.
|
||||
if err := isSafeURL(agentURL); err != nil {
|
||||
return "", fmt.Errorf("invalid workspace URL: %w", err)
|
||||
}
|
||||
|
||||
a2aBody, err := json.Marshal(map[string]interface{}{
|
||||
"jsonrpc": "2.0",
|
||||
"id": uuid.New().String(),
|
||||
@@ -218,36 +204,17 @@ func (h *MCPHandler) toolDelegateTask(ctx context.Context, callerID string, args
|
||||
reqCtx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
|
||||
httpReq, err := http.NewRequestWithContext(reqCtx, "POST", agentURL+"/a2a", bytes.NewReader(a2aBody))
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to create request: %w", err)
|
||||
}
|
||||
httpReq.Header.Set("Content-Type", "application/json")
|
||||
// X-Workspace-ID identifies this caller to the A2A proxy. The /workspaces/:id/a2a
|
||||
// endpoint is intentionally outside WorkspaceAuth (agents do not hold bearer tokens
|
||||
// to peer workspaces). Access control is enforced by CanCommunicate above, which
|
||||
// already validated callerID → targetID before this request is constructed.
|
||||
// callerID was authenticated by WorkspaceAuth on the MCP bridge entry point,
|
||||
// so this header reflects a verified caller identity, not a spoofable value.
|
||||
httpReq.Header.Set("X-Workspace-ID", callerID)
|
||||
|
||||
resp, err := http.DefaultClient.Do(httpReq)
|
||||
status, body, err := h.proxyA2ARequest(reqCtx, targetID, a2aBody, callerID, true)
|
||||
if err != nil {
|
||||
updateMCPDelegationStatus(ctx, h.database, callerID, delegationID, "failed", err.Error())
|
||||
return "", fmt.Errorf("A2A call failed: %w", err)
|
||||
return "", fmt.Errorf("A2A proxy failed: %w", err)
|
||||
}
|
||||
if status < 200 || status >= 300 {
|
||||
updateMCPDelegationStatus(ctx, h.database, callerID, delegationID, "failed", fmt.Sprintf("A2A proxy returned status %d", status))
|
||||
return "", fmt.Errorf("A2A proxy returned status %d", status)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
// A 200/500 from the peer still means the call was dispatched — only
|
||||
// network errors are truly "failed". Status 'dispatched' is correct for
|
||||
// any HTTP response (peer's A2A layer handles the actual processing).
|
||||
updateMCPDelegationStatus(ctx, h.database, callerID, delegationID, "dispatched", "")
|
||||
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to read response: %w", err)
|
||||
}
|
||||
|
||||
return extractA2AText(body), nil
|
||||
}
|
||||
|
||||
@@ -278,24 +245,13 @@ func (h *MCPHandler) toolDelegateTaskAsync(ctx context.Context, callerID string,
|
||||
|
||||
// Fire and forget in a detached goroutine. Use a background context so
|
||||
// the call is not cancelled when the HTTP request completes.
|
||||
// RFC internal#524 Layer 1: globalGoAsync — the detached call reads
|
||||
// db.DB (mcpResolveURL + updateMCPDelegationStatus) and must be
|
||||
// drained by drainTestAsync before any t.Cleanup-driven db.DB swap.
|
||||
// RFC internal#524 Layer 1: globalGoAsync — the detached call reads db.DB
|
||||
// through the platform A2A proxy and must be drained by drainTestAsync
|
||||
// before any t.Cleanup-driven db.DB swap.
|
||||
globalGoAsync(func() {
|
||||
bgCtx, cancel := context.WithTimeout(context.Background(), mcpAsyncCallTimeout)
|
||||
defer cancel()
|
||||
|
||||
agentURL, err := mcpResolveURL(bgCtx, h.database, targetID)
|
||||
if err != nil {
|
||||
log.Printf("MCPHandler.delegate_task_async: resolve URL for %s: %v", targetID, err)
|
||||
return
|
||||
}
|
||||
// SSRF defence: reject private/metadata URLs before making outbound call.
|
||||
if err := isSafeURL(agentURL); err != nil {
|
||||
log.Printf("MCPHandler.delegate_task_async: unsafe URL for %s: %v", targetID, err)
|
||||
return
|
||||
}
|
||||
|
||||
a2aBody, _ := json.Marshal(map[string]interface{}{
|
||||
"jsonrpc": "2.0",
|
||||
"id": delegationID,
|
||||
@@ -309,22 +265,15 @@ func (h *MCPHandler) toolDelegateTaskAsync(ctx context.Context, callerID string,
|
||||
},
|
||||
})
|
||||
|
||||
httpReq, err := http.NewRequestWithContext(bgCtx, "POST", agentURL+"/a2a", bytes.NewReader(a2aBody))
|
||||
if err != nil {
|
||||
log.Printf("MCPHandler.delegate_task_async: create request: %v", err)
|
||||
status, _, err := h.proxyA2ARequest(bgCtx, targetID, a2aBody, callerID, true)
|
||||
if err != nil || status < 200 || status >= 300 {
|
||||
if err != nil {
|
||||
log.Printf("MCPHandler.delegate_task_async: A2A proxy to %s: %v", targetID, err)
|
||||
} else {
|
||||
log.Printf("MCPHandler.delegate_task_async: A2A proxy to %s returned status %d", targetID, status)
|
||||
}
|
||||
return
|
||||
}
|
||||
httpReq.Header.Set("Content-Type", "application/json")
|
||||
httpReq.Header.Set("X-Workspace-ID", callerID)
|
||||
|
||||
resp, err := http.DefaultClient.Do(httpReq)
|
||||
if err != nil {
|
||||
log.Printf("MCPHandler.delegate_task_async: A2A call to %s: %v", targetID, err)
|
||||
return
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
// Drain response so the connection can be reused.
|
||||
_, _ = io.Copy(io.Discard, resp.Body)
|
||||
})
|
||||
|
||||
return fmt.Sprintf(`{"task_id":%q,"status":"dispatched","target_id":%q}`, delegationID, targetID), nil
|
||||
@@ -405,7 +354,6 @@ func (h *MCPHandler) toolSendMessageToUser(ctx context.Context, workspaceID stri
|
||||
return "Message sent.", nil
|
||||
}
|
||||
|
||||
|
||||
func (h *MCPHandler) toolCommitMemory(ctx context.Context, workspaceID string, args map[string]interface{}) (string, error) {
|
||||
// PR-6 (RFC #2728) compat shim: when the v2 plugin is wired
|
||||
// (MEMORY_PLUGIN_URL set), translate legacy scope→namespace and
|
||||
@@ -534,56 +482,6 @@ func (h *MCPHandler) toolRecallMemory(ctx context.Context, workspaceID string, a
|
||||
// Helpers
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// mcpResolveURL returns a routable URL for a workspace's A2A server.
|
||||
//
|
||||
// Resolution order:
|
||||
// 1. Docker-internal URL cache (set by provisioner; correct when platform is in Docker)
|
||||
// 2. Redis URL cache
|
||||
// 3. DB `url` column fallback, with 127.0.0.1→Docker bridge rewrite when in Docker
|
||||
//
|
||||
// SECURITY (F1083 / #1130): all three paths run the returned URL through
|
||||
// validateAgentURL to block SSRF targets (private IPs, loopback, cloud metadata).
|
||||
func mcpResolveURL(ctx context.Context, database *sql.DB, workspaceID string) (string, error) {
|
||||
if platformInDocker {
|
||||
if url, err := db.GetCachedInternalURL(ctx, workspaceID); err == nil && url != "" {
|
||||
if err := validateAgentURL(url); err != nil {
|
||||
return "", fmt.Errorf("workspace %s: forbidden URL from internal cache: %w", workspaceID, err)
|
||||
}
|
||||
return url, nil
|
||||
}
|
||||
}
|
||||
if url, err := db.GetCachedURL(ctx, workspaceID); err == nil && url != "" {
|
||||
if platformInDocker && strings.HasPrefix(url, "http://127.0.0.1:") {
|
||||
return provisioner.InternalURL(workspaceID), nil
|
||||
}
|
||||
if err := validateAgentURL(url); err != nil {
|
||||
return "", fmt.Errorf("workspace %s: forbidden URL from Redis cache: %w", workspaceID, err)
|
||||
}
|
||||
return url, nil
|
||||
}
|
||||
|
||||
var urlStr sql.NullString
|
||||
var status string
|
||||
if err := database.QueryRowContext(ctx,
|
||||
`SELECT url, status FROM workspaces WHERE id = $1`, workspaceID,
|
||||
).Scan(&urlStr, &status); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return "", fmt.Errorf("workspace %s not found", workspaceID)
|
||||
}
|
||||
return "", fmt.Errorf("workspace lookup failed: %w", err)
|
||||
}
|
||||
if !urlStr.Valid || urlStr.String == "" {
|
||||
return "", fmt.Errorf("workspace %s has no URL (status: %s)", workspaceID, status)
|
||||
}
|
||||
if platformInDocker && strings.HasPrefix(urlStr.String, "http://127.0.0.1:") {
|
||||
return provisioner.InternalURL(workspaceID), nil
|
||||
}
|
||||
if err := validateAgentURL(urlStr.String); err != nil {
|
||||
return "", fmt.Errorf("workspace %s: forbidden URL from DB: %w", workspaceID, err)
|
||||
}
|
||||
return urlStr.String, nil
|
||||
}
|
||||
|
||||
// extractA2AText extracts human-readable text from an A2A JSON-RPC response body.
|
||||
// Falls back to the raw JSON when no text part can be found.
|
||||
func extractA2AText(body []byte) string {
|
||||
@@ -632,4 +530,3 @@ func extractA2AText(body []byte) string {
|
||||
b, _ := json.Marshal(result)
|
||||
return string(b)
|
||||
}
|
||||
|
||||
|
||||
@@ -112,7 +112,7 @@ func (h *RegistryHandler) SetQueueDrainFunc(f QueueDrainFunc) {
|
||||
// Go's net.ParseIP.To4() before Contains() runs, so the IPv4 rules above
|
||||
// catch those without a separate entry.
|
||||
//
|
||||
// F1083/#1130 (SSRF on mcpResolveURL / a2a_proxy resolveAgentURL): in
|
||||
// F1083/#1130 (SSRF on direct A2A URL resolution): in
|
||||
// addition to blocking IP literals, DNS names are now resolved and each
|
||||
// returned IP is checked against the blocklist. This closes the gap where
|
||||
// an attacker could register agent.example.com pointing to 169.254.169.254.
|
||||
|
||||
@@ -397,6 +397,8 @@ func Setup(hub *ws.Hub, broadcaster *events.Broadcaster, prov *provisioner.Provi
|
||||
wsAuth.GET("/tokens", tokh.List)
|
||||
wsAuth.POST("/tokens", tokh.Create)
|
||||
wsAuth.DELETE("/tokens/:tokenId", tokh.Revoke)
|
||||
adminTokH := handlers.NewAdminWorkspaceTokenHandler()
|
||||
r.POST("/admin/workspaces/:id/tokens", middleware.AdminAuth(db.DB), adminTokH.Create)
|
||||
|
||||
// Memory
|
||||
memh := handlers.NewMemoryHandler()
|
||||
|
||||
Reference in New Issue
Block a user