From 06682f33c635b97b58984c5d3db9dc1926f9e75b Mon Sep 17 00:00:00 2001 From: Renzo F <45401804+rennf93@users.noreply.github.com> Date: Fri, 5 Jun 2026 16:35:22 +0200 Subject: [PATCH] Fix: agent workflow hardening (#70) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(gateway): push the branch before QA handoff so reviewers see the latest commits The commit content tool commits locally without pushing; only open_pr pushed the branch. On the first submission that was fine, but a fix committed while addressing needs_revision never reached origin (open_pr is skipped once the PR exists), so QA — which reviews the remote PR branch — re-reviewed the stale remote and re-failed the task on every cycle, a loop that never converged. i_am_done now pushes the task branch (idempotent; a no-op when nothing is unpushed) as part of the shared submit gate, covering both the normal and resume-from-verifying paths. A push failure blocks the handoff with a clear remediation rather than parking the task in awaiting_qa with commits that exist only in the developer's local workspace. * fix(orchestrator): don't reap a stale claim while the agent's container is alive The stale-claim reaper released any claimed/in_progress task whose last_heartbeat_at exceeded the TTL. The heartbeat only updates on certain gateway calls, so a developer deep in a long edit/test cycle outran the TTL and had its claim reaped mid-work — churning the task and risking a double spawn against the still-running container. The reaper now skips a task whose assignee still holds a live (ACTIVE) agent instance, trusting container liveness — the ground truth — over the heartbeat proxy. The check is defensive on missing fields so a heartbeat-only caller (and the reaper's existing unit tests) behave exactly as before. * fix(gateway): refuse to unblock a task while a dependency is unfinished A PM unblock on a dependency-gated task moved it straight to in_progress, overriding the dependency — letting a dependent proceed without its upstream's work (e.g. a frontend task built before its UX design lands). A dependency block is meant to clear on its own via _unblock_dependents the moment the upstream reaches a terminal state. unblock now refuses while any dependency is still non-terminal, returning a clear remediation that the block resolves automatically. Manual unblock remains available for genuine, non-dependency blockers. * fix(gateway): release a dependency-blocked claim to pending instead of looping A task that reached claimed/in_progress with an unfinished dependency was left in that state when the claim guard rejected, so the orchestrator's respawn loop kept reviving its assignee — which could make no progress — burning work for nothing. The claim guard now releases such a task back to pending. claimed -> blocked is not a legal transition, so pending — held by the dispatch dependency filter — is the lifecycle-correct resting state: the respawn loop ignores pending tasks, and _unblock_dependents re-dispatches it once the upstream reaches a terminal state. release_dependency_blocked_claim shares a _force_unclaim_to_pending core with unclaim_for_reaper so both record a truthful work-session abandon reason. * feat(security): warn at startup in header-trust mode + document the auth posture When ROBOCO_AGENT_AUTH_REQUIRED is not enabled the API accepts the X-Agent-Id / X-Agent-Role headers without a signed token, so any client that can reach it may act as any role (including 'ceo'). The API now logs a clear warning at startup in this mode, and the README gains a Security section documenting the auth posture and how to harden it. Acceptable only on a trusted private network — do not expose the API to untrusted networks. * fix(workspace): scope the refresh fetch to current + default branch ensure_workspace's healthy short-circuit ran an all-refs 'git fetch origin' to keep every origin/ ref current. On a monorepo with many accumulated feature/* branches that exceeds the refresh timeout, the fetch silently fails, and the workspace keeps a stale base — so an agent builds on an out-of-date branch. The refresh now fetches only the workspace's current branch and the repo's default branch (resolved via origin/HEAD), with --no-tags --prune: it transfers near-nothing and can't time out. Readers need their own branch and the default; the integration branch is refreshed at branch-creation time. * fix(git): refresh a dependency-blocked task's branch off the current integration tip A cross-cell dependent (e.g. a frontend task waiting on the UX design) was branched off a base captured before its upstream merged into the integration branch, and the branch was never re-synced — so the agent built on a stale snapshot with none of the upstream's work. Two changes close the gap: - release_dependency_blocked_claim now clears branch_name, so the re-claim (after the dependency clears) re-runs branch creation. - create_branch, when the branch is already on disk with no commits of its own, resets it onto the freshly-pulled base — the dependent now builds on the current integration tip. A branch carrying real commits is left untouched, so no work is discarded; the cell->leaf cascade carries the upstream down to the dev branch automatically. * refactor(gateway): drop the sibling-sequence claim guard Sibling sequence no longer gates a claim. Cross-cell ordering is enforced by task dependencies — a cell task that depends on another is held until its upstream reaches a terminal state, a stronger, status-aware gate than the sequence-number check. That check was dormant in practice anyway: every fan-out child carries sequence 0, on which the guard short-circuited. `sequence` stays a sibling-ordering / dispatch-priority field (list_pending ordering and the panel). Removes sibling_sequence_guard and its _earlier_blocking_sibling helper, the now-unused skip_sequence parameter threaded through the claim verbs, and the sibling fetch that fed it. * feat(gateway): sort a cross-cell dependent after its upstream When the frontend cell task is wired to depend on its UX/UI sibling, set its sequence to the upstream's sequence + 1 so it sorts after the design it waits on — list_pending ordering and the panel now show UX ahead of the implementation it gates, in either delegation order. Adds TaskService.set_sequence (the sibling-ordering field is a service write; it carries no claim-gating semantics — dependencies gate claims). * feat(gateway): make the backend cell depend on UX too UX/UI design defines the screens and API contracts both implementation cells build against, so the backend cell — not just the frontend — waits on the UX/UI cell task in a product fan-out and sorts after it. Wires in either delegation order: a backend task delegated after UX gets the dependency directly; a UX task delegated after a still-pending backend sibling retro-wires it. Mirrors the existing frontend wiring (_depend_backend_on_ux and _depend_pending_backends_on_ux). Backend is held by the same dependency gate, so it costs no extra dispatch churn. * fix(websocket): forward notification acks instead of logging them incomplete The bridge handler serves both notification.sent and notification.acked, but acked events carry `agent_id` (the acking agent) rather than `recipient_id`, so every acknowledgement tripped the missing-field guard and logged "Incomplete notification event" instead of reaching the panel. Accept either field as the recipient. * feat(api): hint the full UUID when a truncated task id fails validation Agents copy the 8-character task prefix the system shows them (the commit prefix, task summaries) and send it as task_id, which fails UUID validation with an opaque "invalid length" 422 and wastes a call. The request-validation handler now detects a task_id UUID error and attaches a `remediate` hint telling the agent to retry with the full 36-character UUID from its task envelope. * fix(audit): record the blocked transition when a task is escalated Escalation sets a task to blocked by writing task.status directly, which bypassed the validated transition helper and so never emitted a task.blocked audit row — the lifecycle moved but the Auditor saw nothing. Extract the audit emit from the central transition helper into _emit_status_transition_audit and call it from the escalate path, capturing the prior status and outgoing owner before reassignment so the row is attributed correctly. * fix(docs): stop doubling the docs path so design specs index into RAG The documenter sometimes hands a doc path already rooted at docs/, and joining it onto DOCS_BASE_PATH (/app/docs) produced /app/docs/docs/..., so the file was never found and the spec never indexed — the frontend cell could not retrieve the UX design over RAG. Normalize the path before joining: trust an absolute path, otherwise strip a single redundant leading docs/ segment. * feat(security): let the control panel authenticate in secure mode With ROBOCO_AGENT_AUTH_REQUIRED=true every request must carry a valid HMAC token, which locked the human control panel out — it sends role headers but no token. nginx, the only trusted hop between the browser and the API, now injects the CEO token on /api and /ws, so the browser never holds the signing secret. The injected value is just the existing per-agent token issued for the CEO identity (issue_panel_token), so the token-verification path is unchanged. An empty value (dev/header-trust mode) renders to no header. `make panel-token` prints the value; set it as ROBOCO_PANEL_AGENT_TOKEN in .env before enabling secure mode. .env.example and the README Security section document the flow. * chore(compose): consolidate the two compose files into one docker-compose.yml and docker-compose.yaml had diverged: .yml — the file Docker actually uses — carried ROBOCO_PUBLIC_BASE_URL but was missing the /app/manifests bind-mount, while .yaml had the manifests mount but not the base URL. Merge the union into docker-compose.yml and delete the duplicate so there is one source of truth and no "multiple config files" warning. This activates the manifests mount in the deployed file: without it the orchestrator writes per-agent tool manifests to its ephemeral container fs, they never reach the host for the daemon to bind-mount, and agents fall back to all-verbs registration. Drop the stale .yaml reference from the config.py docstring, the labeler, and the CI path filters. --------- Co-authored-by: Renn F --- .env.example | 13 + .github/labeler.yml | 1 - .github/workflows/ci.yml | 2 - Makefile | 8 + README.md | 26 ++ docker-compose.yaml | 316 ------------------ docker-compose.yml | 16 +- docker/nginx.conf | 6 + roboco/agents_config.py | 16 +- roboco/api/app.py | 11 + roboco/api/middleware.py | 39 ++- roboco/api/websocket_bridge.py | 6 +- roboco/config.py | 2 +- roboco/runtime/orchestrator.py | 27 +- .../services/gateway/choreographer/_impl.py | 164 ++++++--- .../gateway/choreographer/_protocol.py | 1 - roboco/services/gateway/choreographer/doc.py | 1 - roboco/services/gateway/choreographer/qa.py | 1 - roboco/services/gateway/claim_guards.py | 51 --- roboco/services/git.py | 36 ++ roboco/services/task.py | 164 +++++++-- roboco/services/workspace.py | 37 +- .../test_lifecycle_consumer_parity.py | 4 +- ..._subtask_inherits_cross_cell_dependency.py | 153 ++++++++- .../test_full_lifecycle_real_db.py | 4 + tests/integration/test_lifecycle_real_db.py | 4 + ...e_assigned_dev_held_on_unmet_dependency.py | 43 +++ tests/unit/api/test_app.py | 60 +++- tests/unit/api/test_middleware.py | 62 ++++ tests/unit/api/test_websocket_bridge.py | 24 ++ .../test_choreographer_claim_guards.py | 151 +-------- tests/unit/gateway/test_choreographer_dev.py | 90 +++++ tests/unit/gateway/test_choreographer_pm.py | 29 ++ .../unit/gateway/test_claim_guards_direct.py | 39 --- tests/unit/runtime/test_stale_claim_reaper.py | 52 ++- .../services/test_escalation_board_guard.py | 46 ++- tests/unit/services/test_git.py | 118 +++++++ tests/unit/services/test_task.py | 32 ++ tests/unit/services/test_workspace_refresh.py | 63 +++- tests/unit/test_agents_config.py | 35 ++ 40 files changed, 1300 insertions(+), 653 deletions(-) delete mode 100644 docker-compose.yaml delete mode 100644 tests/unit/gateway/test_claim_guards_direct.py diff --git a/.env.example b/.env.example index e928de65..66fc94af 100644 --- a/.env.example +++ b/.env.example @@ -78,6 +78,19 @@ ROBOCO_QDRANT_PORT=6333 # ============================================================================= # Security # ============================================================================= +# Agent auth: HMAC secret that signs X-Agent-Token. REQUIRED for docker compose. +# Generate with: python -c 'import secrets; print(secrets.token_hex(32))' +ROBOCO_AGENT_AUTH_SECRET= + +# Secure mode. On a trusted LAN you can leave this false (header-trust mode). +# Set true to require every request to carry a valid token so an agent cannot +# spoof another agent's role. When true you MUST also set ROBOCO_PANEL_AGENT_TOKEN. +ROBOCO_AGENT_AUTH_REQUIRED=false + +# The control panel's CEO token, injected by nginx in secure mode so the human +# UI keeps working without the browser holding the signing secret. Generate it +# (after setting ROBOCO_AGENT_AUTH_SECRET above) with: make panel-token +ROBOCO_PANEL_AGENT_TOKEN= # ============================================================================= # CORS (comma-separated origins) diff --git a/.github/labeler.yml b/.github/labeler.yml index 68a7f1dd..2dccf2bf 100644 --- a/.github/labeler.yml +++ b/.github/labeler.yml @@ -76,7 +76,6 @@ build: - Makefile - docker/** - docker-compose.yml - - docker-compose.yaml dependencies: - changed-files: diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2bec6055..0c57a96a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,7 +12,6 @@ on: - 'scripts/**' - 'docker/**' - 'docker-compose.yml' - - 'docker-compose.yaml' - 'Makefile' - 'pyproject.toml' - 'uv.lock' @@ -30,7 +29,6 @@ on: - 'scripts/**' - 'docker/**' - 'docker-compose.yml' - - 'docker-compose.yaml' - 'Makefile' - 'pyproject.toml' - 'uv.lock' diff --git a/Makefile b/Makefile index cf77f0d0..3b874334 100644 --- a/Makefile +++ b/Makefile @@ -386,6 +386,13 @@ prune: clean: @find . | grep -E "(__pycache__|\.pyc|\.pyo|\.pytest_cache|\.ruff_cache|\.mypy_cache)" | xargs rm -rf +# Security +.PHONY: panel-token +panel-token: + @SECRET="$$(grep -E '^ROBOCO_AGENT_AUTH_SECRET=' .env 2>/dev/null | head -1 | cut -d= -f2-)"; \ + ROBOCO_AGENT_AUTH_SECRET="$${SECRET:-$$ROBOCO_AGENT_AUTH_SECRET}" \ + uv run python -c "import sys; from roboco.agents_config import issue_panel_token; tok = issue_panel_token(); print(tok) if tok != 'UNSIGNED' else sys.exit('ERROR: ROBOCO_AGENT_AUTH_SECRET not set (in .env or environment) - the panel token would be unsigned')" + # Help .PHONY: help help: @@ -422,6 +429,7 @@ help: @echo " make quality - Run all quality checks" @echo " make security - Run security checks (bandit, safety, pip-audit)" @echo " make check-all - Run ALL checks" + @echo " make panel-token - Print the panel's CEO token for secure mode" @echo "" @echo "Testing:" @echo " make test - Run tests (Python $(DEFAULT_PYTHON))" diff --git a/README.md b/README.md index c71bf1bc..c6278115 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,32 @@ uv run mypy roboco/ - [x] Frontend panel (vendored under `panel/`, served through nginx on :3000) - [ ] Full agent autonomy testing +## Security + +> [!IMPORTANT] +> **Do not expose RoboCo to the public internet as-is.** It is designed to run +> on a trusted private network (homelab / LAN). + +**Agent authentication.** Requests identify the caller with `X-Agent-Id` / +`X-Agent-Role` headers. The orchestrator issues each spawned agent an HMAC token +(`X-Agent-Token`, signed with `ROBOCO_AGENT_AUTH_SECRET`) that binds its id, role +and team. Token enforcement is gated by `ROBOCO_AGENT_AUTH_REQUIRED`: + +- **`ROBOCO_AGENT_AUTH_REQUIRED` unset/false (default):** *header-trust mode* — + the role headers are accepted without a token, so any client that can reach the + API may claim any role (including `ceo`). The API logs a warning at startup in + this mode. Acceptable only on a trusted network. +- **`ROBOCO_AGENT_AUTH_REQUIRED=true`:** every request must carry a valid token; + an agent cannot spoof another agent's role. The control panel keeps working + because **nginx** — the only trusted hop between the browser and the API — + injects the CEO token (`X-Agent-Token`) on `/api` and `/ws`, so the browser + never holds the signing secret. Generate that token with `make panel-token` + and set it as `ROBOCO_PANEL_AGENT_TOKEN` in `.env` before enabling secure mode. + +**Secrets** (the Fernet `ROBOCO_ENCRYPTION_KEY`, GitHub PATs) live encrypted in +the database and in gitignored env files — never in the repo. Per-project git +tokens are Fernet-encrypted at rest and never returned by the API. + ## License Copyright (c) 2026 Renzo Franceschini diff --git a/docker-compose.yaml b/docker-compose.yaml deleted file mode 100644 index 5e44764a..00000000 --- a/docker-compose.yaml +++ /dev/null @@ -1,316 +0,0 @@ -services: - # ========================================================================== - # PostgreSQL - Primary Database with pgvector for RAG - # ========================================================================== - postgres: - image: pgvector/pgvector:pg16 - container_name: roboco-postgres - restart: unless-stopped - environment: - POSTGRES_USER: roboco - POSTGRES_PASSWORD: roboco - POSTGRES_DB: roboco - ports: - - "15432:5432" - volumes: - - ${ROBOCO_DATA_DIR:-./data}/postgres:/var/lib/postgresql/data - healthcheck: - test: ["CMD-SHELL", "pg_isready -U roboco -d roboco"] - interval: 10s - timeout: 5s - retries: 5 - - # ========================================================================== - # Redis - Cache, Sessions, Event Bus - # ========================================================================== - redis: - image: redis:8-alpine - container_name: roboco-redis - restart: unless-stopped - command: redis-server --appendonly yes - ports: - - "16379:6379" - volumes: - - ${ROBOCO_DATA_DIR:-./data}/redis:/data - healthcheck: - test: ["CMD", "redis-cli", "ping"] - interval: 10s - timeout: 5s - retries: 5 - - # ========================================================================== - # Ollama - Local LLM and Embedding Server - # ========================================================================== - ollama: - image: ollama/ollama:latest - container_name: roboco-ollama - restart: unless-stopped - environment: - OLLAMA_API_KEY: ${OLLAMA_API_KEY} - ports: - - "11435:11434" - volumes: - - ${ROBOCO_DATA_DIR:-./data}/ollama:/root/.ollama - healthcheck: - # Use ollama CLI (guaranteed available) to check if server is responding - test: ["CMD", "ollama", "list"] - interval: 10s - timeout: 5s - retries: 5 - start_period: 10s - - # Ollama model puller - pulls required models on startup - # Uses streaming curl to wait for full model download - ollama-init: - image: curlimages/curl:latest - container_name: roboco-ollama-init - depends_on: - ollama: - condition: service_healthy - restart: "no" - entrypoint: ["/bin/sh", "-c"] - command: - - | - set -e - echo "=== Pulling embedding model (qwen3-embedding:0.6b) ===" - # Ollama /api/pull streams JSON lines until complete - consume full stream - # Note: $$ escapes $ for docker-compose variable substitution - curl -sN http://ollama:11434/api/pull -d '{"name":"qwen3-embedding:0.6b"}' | while read -r line; do - status=$$(echo "$$line" | grep -o '"status":"[^"]*"' | cut -d'"' -f4) - [ -n "$$status" ] && echo " $$status" - done - echo "=== Pulling LLM model (glm-5:cloud) ===" - curl -sN http://ollama:11434/api/pull -d '{"name":"glm-5:cloud"}' | while read -r line; do - status=$$(echo "$$line" | grep -o '"status":"[^"]*"' | cut -d'"' -f4) - [ -n "$$status" ] && echo " $$status" - done - echo "=== Verifying models are available ===" - curl -sf http://ollama:11434/api/tags | grep -q "qwen3-embedding" && echo " qwen3-embedding: OK" - curl -sf http://ollama:11434/api/tags | grep -q "glm-5" && echo " glm-5: OK" - echo "=== All models ready! ===" - - # ========================================================================== - # Agent Base Image Builder (specialized images built on-demand by orchestrator) - # ========================================================================== - agent-base-image: - build: - context: . - dockerfile: docker/agent-base.Dockerfile - image: roboco-agent-base - container_name: roboco-agent-base-builder - entrypoint: ["/bin/sh", "-c", "echo 'Agent base image built successfully'"] - restart: "no" - - # ========================================================================== - # Agent PM Image Builder (specialized image built on-demand by orchestrator) - # ========================================================================== - agent-pm-image: - build: - context: . - dockerfile: docker/agent-pm.Dockerfile - image: roboco-agent-pm - entrypoint: ["/bin/sh", "-c", "echo 'Agent PM image built'"] - restart: "no" - depends_on: - - agent-base-image - - # ========================================================================== - # Agent Backend Dev Image Builder (specialized image built on-demand by orchestrator) - # ========================================================================== - agent-dev-be-image: - build: - context: . - dockerfile: docker/agent-dev-be.Dockerfile - image: roboco-agent-dev-be - entrypoint: ["/bin/sh", "-c", "echo 'Agent Backend Dev image built'"] - restart: "no" - depends_on: - - agent-base-image - - # ========================================================================== - # Agent Frontend Dev Image Builder (specialized image built on-demand by orchestrator) - # ========================================================================== - agent-dev-fe-image: - build: - context: . - dockerfile: docker/agent-dev-fe.Dockerfile - image: roboco-agent-dev-fe - entrypoint: ["/bin/sh", "-c", "echo 'Agent Frontend Dev image built'"] - restart: "no" - depends_on: - - agent-base-image - - # ========================================================================== - # Agent Backend QA Image Builder (specialized image built on-demand by orchestrator) - # ========================================================================== - agent-qa-be-image: - build: - context: . - dockerfile: docker/agent-qa-be.Dockerfile - image: roboco-agent-qa-be - entrypoint: ["/bin/sh", "-c", 'echo "Agent Backend QA image built"'] - restart: "no" - depends_on: - - agent-base-image - - # ========================================================================== - # Agent Frontend QA Image Builder (specialized image built on-demand by orchestrator) - # ========================================================================== - agent-qa-fe-image: - build: - context: . - dockerfile: docker/agent-qa-fe.Dockerfile - image: roboco-agent-qa-fe - entrypoint: ["/bin/sh", "-c", 'echo "Agent Frontend QA image built"'] - restart: "no" - depends_on: - - agent-base-image - - # ========================================================================== - # Agent UX/UI Dev Image Builder (specialized image built on-demand by orchestrator) - # ========================================================================== - agent-ux-image: - build: - context: . - dockerfile: docker/agent-ux.Dockerfile - image: roboco-agent-ux - entrypoint: ["/bin/sh", "-c", 'echo "Agent UX/UI image built"'] - restart: "no" - depends_on: - - agent-base-image - - # ========================================================================== - # Agent Documenter Image Builder (specialized image built on-demand by orchestrator) - # ========================================================================== - agent-doc-image: - build: - context: . - dockerfile: docker/agent-doc.Dockerfile - image: roboco-agent-doc - entrypoint: ["/bin/sh", "-c", 'echo "Agent Documenter image built"'] - restart: "no" - depends_on: - - agent-base-image - - # ========================================================================== - # Orchestrator - API Server + Agent Spawner - # ========================================================================== - orchestrator: - build: - context: . - dockerfile: docker/orchestrator.Dockerfile - image: roboco-orchestrator - container_name: roboco-orchestrator - restart: unless-stopped - ports: - - "8000:8000" - environment: - # Database (use container name, not localhost) - ROBOCO_DATABASE_HOST: roboco-postgres - ROBOCO_DATABASE_PORT: 5432 - ROBOCO_DATABASE_USER: roboco - ROBOCO_DATABASE_PASSWORD: roboco - ROBOCO_DATABASE_NAME: roboco - # Redis (use container name) - ROBOCO_REDIS_HOST: roboco-redis - ROBOCO_REDIS_PORT: 6379 - # API - ROBOCO_HOST: 0.0.0.0 - ROBOCO_PORT: 8000 - ROBOCO_ENCRYPTION_KEY: ${ROBOCO_ENCRYPTION_KEY:?ROBOCO_ENCRYPTION_KEY is required} - # HMAC secret for agent auth tokens. Orchestrator signs tokens - # per-agent at spawn; API middleware verifies them. Generate with: - # python -c 'import secrets; print(secrets.token_hex(32))' - ROBOCO_AGENT_AUTH_SECRET: ${ROBOCO_AGENT_AUTH_SECRET:?ROBOCO_AGENT_AUTH_SECRET is required} - # Set to "true" to require tokens on every API call (fail-closed). - # Leave unset/false during rollout so the panel + curl still work. - ROBOCO_AGENT_AUTH_REQUIRED: ${ROBOCO_AGENT_AUTH_REQUIRED:-false} - # Ollama (use container name) - ROBOCO_LOCAL_LLM_BASE_URL: http://roboco-ollama:11434/v1 - ROBOCO_LOCAL_LLM_MODEL: glm-5:cloud - ROBOCO_DEFAULT_EMBEDDING_MODEL: qwen3-embedding:0.6b - ROBOCO_OLLAMA_BASE_URL: http://roboco-ollama:11434 - # Host paths for spawning agent containers (required for Docker-in-Docker) - # IMPORTANT: These must be ABSOLUTE paths on the host filesystem - ROBOCO_HOST_PROJECT_DIR: ${ROBOCO_HOST_PROJECT_DIR:-/volume1/roboco} - ROBOCO_HOST_CLAUDE_DIR: ${ROBOCO_HOST_CLAUDE_DIR:-/home/renzof/.claude} - ROBOCO_HOST_DATA_DIR: ${ROBOCO_HOST_DATA_DIR:-/volume1/roboco/data} - # Production environment selects structlog's JSONRenderer (machine- - # parseable logs) over the dev ConsoleRenderer. - ROBOCO_ENVIRONMENT: production - volumes: - # Docker socket - allows spawning agent containers - - /var/run/docker.sock:/var/run/docker.sock - # Claude Code auth - mount your ~/.claude directory - - ${CLAUDE_AUTH_DIR:-/home/renzof/.claude}:/root/.claude - # Shared config directory for MCP configs (writable) - - ${ROBOCO_DATA_DIR:-./data}/mcp-configs:/app/mcp-configs - # Generated prompts directory - composed at runtime from layers - - ${ROBOCO_DATA_DIR:-./data}/prompts-generated:/app/prompts-generated - # Per-agent Claude settings (generated at spawn time) - - ${ROBOCO_DATA_DIR:-./data}/agent-settings:/app/agent-settings - # Per-agent SessionStart briefings (pre-rendered task context) - - ${ROBOCO_DATA_DIR:-./data}/briefings:/app/briefings - # Per-agent spawn manifests (role-scoped tool list) — written by the - # orchestrator, bind-mounted into each agent container as - # /app/tool-manifest.json. Without this mount the file is invisible - # to the Docker daemon and agents fall back to all-verbs registration. - - ${ROBOCO_DATA_DIR:-./data}/manifests:/app/manifests - # Agent workspaces (git clones) - persisted across restarts - - ${ROBOCO_DATA_DIR:-./data}/workspaces:/data/workspaces - # Persistent logs — survive `docker compose down/up`. Orchestrator and - # each spawned agent write structured logs here so we can audit past - # runs instead of relying on ephemeral `docker logs`. - - ${ROBOCO_DATA_DIR:-./data}/logs:/data/logs - depends_on: - postgres: - condition: service_healthy - redis: - condition: service_healthy - ollama: - condition: service_healthy - ollama-init: - condition: service_completed_successfully - agent-base-image: - condition: service_completed_successfully - # Default agents to spawn (override in .env or command line) - # command: ["--spawn", "main-pm", "be-dev-1", "be-qa"] - - # ========================================================================== - # Next.js Control Panel (Frontend) - # ========================================================================== - # Not exposed directly - nginx is the single entry point (port 3000). - # API/WS traffic is proxied to the orchestrator, everything else to panel. - panel: - build: - context: . - dockerfile: docker/panel.Dockerfile - image: roboco-panel - container_name: roboco-panel - restart: unless-stopped - expose: - - "3000" - depends_on: - - orchestrator - - # ========================================================================== - # Nginx - Reverse proxy fronting panel + orchestrator - # ========================================================================== - # Single entry point on port 3000 so the browser hits one origin and we - # don't need CORS. /api/* and /ws/* go to the orchestrator, everything - # else goes to the Next.js panel. - nginx: - image: nginx:alpine - container_name: roboco-nginx - restart: unless-stopped - ports: - - "3000:80" - volumes: - - ./docker/nginx.conf:/etc/nginx/conf.d/default.conf:ro - depends_on: - - panel - - orchestrator - -networks: - default: - name: roboco_default diff --git a/docker-compose.yml b/docker-compose.yml index 56abc2b1..26b15055 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -261,6 +261,12 @@ services: - ${ROBOCO_DATA_DIR:-./data}/logs:/data/logs # Per-agent SessionStart briefings (pre-rendered task context) - ${ROBOCO_DATA_DIR:-./data}/briefings:/app/briefings + # Per-agent spawn manifests (role-scoped tool list) — written by the + # orchestrator to /app/manifests, bind-mounted into each agent container + # as /app/tool-manifest.json. Without this mount the file is written to + # the orchestrator's ephemeral fs, never reaches the host, and agents + # fall back to all-verbs registration. + - ${ROBOCO_DATA_DIR:-./data}/manifests:/app/manifests depends_on: postgres: condition: service_healthy @@ -304,8 +310,16 @@ services: restart: unless-stopped ports: - "3000:80" + environment: + # Rendered into the proxy config by the nginx image's envsubst + # entrypoint so the human panel authenticates in secure mode. The + # filter limits substitution to ROBOCO_* vars, leaving nginx's own + # $host / $remote_addr runtime variables untouched. Get the value + # with `make panel-token`. + ROBOCO_PANEL_AGENT_TOKEN: ${ROBOCO_PANEL_AGENT_TOKEN:-} + NGINX_ENVSUBST_FILTER: "^ROBOCO_" volumes: - - ./docker/nginx.conf:/etc/nginx/conf.d/default.conf:ro + - ./docker/nginx.conf:/etc/nginx/templates/default.conf.template:ro depends_on: - panel - orchestrator diff --git a/docker/nginx.conf b/docker/nginx.conf index 30e2ce61..f7488616 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -35,6 +35,11 @@ server { proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; + # The browser never holds the signing secret. nginx (the only trusted + # hop for the human panel) attaches the CEO token so secure mode + # (ROBOCO_AGENT_AUTH_REQUIRED=true) does not lock the panel out. Empty + # when unset, in which case nginx sends no header (dev/header-trust). + proxy_set_header X-Agent-Token "${ROBOCO_PANEL_AGENT_TOKEN}"; } # WebSocket requests -> orchestrator @@ -46,6 +51,7 @@ server { proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Agent-Token "${ROBOCO_PANEL_AGENT_TOKEN}"; proxy_read_timeout 86400; } diff --git a/roboco/agents_config.py b/roboco/agents_config.py index 04ecdc83..6a07a636 100644 --- a/roboco/agents_config.py +++ b/roboco/agents_config.py @@ -33,7 +33,7 @@ from typing import Final from roboco.foundation import identity as _foundation from roboco.foundation.policy import communications as _comms from roboco.models.base import NotificationPriority, NotificationType -from roboco.seeds.initial_data import AGENT_UUIDS +from roboco.seeds.initial_data import AGENT_UUIDS, CEO_AGENT_ID # Env var containing the HMAC secret used to sign agent auth tokens. # Must be set in orchestrator + API container environments; if missing, @@ -91,6 +91,20 @@ def verify_agent_token(token: str, agent_id: str, role: str, team: str = "") -> return hmac.compare_digest(expected, token) +def issue_panel_token() -> str: + """Mint the token the control panel presents to act as the CEO. + + The panel calls the API as the CEO identity — ``X-Agent-Id`` = the CEO + UUID, ``X-Agent-Role`` = ``ceo``, and no team header — so the token is + signed for exactly those values (empty team). In secure mode nginx injects + it as ``X-Agent-Token`` so the browser never holds the signing secret; + this is just the existing per-agent token issued for the CEO identity, so + the verification path is unchanged. Returns ``UNSIGNED`` when the secret is + unset (same fail-closed contract as ``issue_agent_token``). + """ + return issue_agent_token(CEO_AGENT_ID, "ceo", "") + + # Reverse mapping: UUID -> slug (computed from seeds) _UUID_TO_SLUG: Final[dict[str, str]] = { uuid: slug for slug, uuid in AGENT_UUIDS.items() diff --git a/roboco/api/app.py b/roboco/api/app.py index 6d6fb571..9f28eaa5 100644 --- a/roboco/api/app.py +++ b/roboco/api/app.py @@ -11,6 +11,7 @@ from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware +from roboco.api.deps import _auth_required from roboco.api.middleware import setup_middleware from roboco.api.routes.a2a import router as a2a_router from roboco.api.routes.a2a import wellknown_router as a2a_wellknown_router @@ -76,6 +77,16 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: environment=settings.environment, ) + if not _auth_required(): + logger.warning( + "Agent auth is in HEADER-TRUST mode (ROBOCO_AGENT_AUTH_REQUIRED is " + "not set to true): the API accepts X-Agent-Id / X-Agent-Role without " + "verifying a signed token, so any client that can reach it may act as " + "any role, including 'ceo'. Acceptable only on a trusted private " + "network. Set ROBOCO_AGENT_AUTH_REQUIRED=true and do NOT expose this " + "API to untrusted networks.", + ) + # Startup: apply Alembic migrations (+ create_all fallback for fresh DBs). # init_db runs on every environment now — migrations are idempotent via # alembic_version, and this is the only way new schema (e.g. enum value diff --git a/roboco/api/middleware.py b/roboco/api/middleware.py index 7f7fa2c6..47f6a572 100644 --- a/roboco/api/middleware.py +++ b/roboco/api/middleware.py @@ -6,8 +6,8 @@ Request/response middleware for logging, error handling, and correlation IDs. import time import uuid -from collections.abc import Callable -from typing import cast +from collections.abc import Callable, Sequence +from typing import Any, cast import structlog from fastapi import FastAPI, HTTPException, Request, Response @@ -303,6 +303,29 @@ async def http_exception_handler(request: Request, exc: Exception) -> JSONRespon # ============================================================================= +def _uuid_field_remediation(errors: Sequence[Any]) -> str | None: + """Spell out the fix when a truncated id is sent where a UUID is required. + + Agents routinely copy the 8-character task prefix the system shows them + (e.g. the ``[cee99ecc]`` commit prefix) and send it as ``task_id``, which + fails UUID validation with an opaque "invalid length" message and wastes a + call. Detect that case and hand back an actionable remediation instead. + """ + for err in errors: + if not isinstance(err, dict): + continue + loc = err.get("loc") or () + field = loc[-1] if loc else None + if field == "task_id" and "uuid" in str(err.get("type", "")).lower(): + return ( + "Use the FULL 36-character task UUID, not the 8-character short " + "form shown in commit prefixes or summaries. The full id is in " + "the `task_id` field of the envelope returned by give_me_work " + "or your most recent verb." + ) + return None + + async def request_validation_handler(request: Request, exc: Exception) -> JSONResponse: """Log the rejected body before returning the standard 422 response. @@ -310,19 +333,27 @@ async def request_validation_handler(request: Request, exc: Exception) -> JSONRe nothing lands in server logs. During smoke tests this leaves us blind to which field actually broke. Log the body + the per-field errors so the next 422 is debuggable in one log scan. + + When the failure is a truncated ``task_id`` (the recurring agent mistake), + add a ``remediate`` hint so the agent knows to retry with the full UUID. """ rve = cast("RequestValidationError", exc) body = rve.body if isinstance(rve.body, str | bytes | dict | list) else None + errors = rve.errors() logger.warning( "Request validation failed", path=request.url.path, method=request.method, body=body, - errors=rve.errors(), + errors=errors, ) + content: dict[str, Any] = {"detail": errors, "body": body} + remediate = _uuid_field_remediation(errors) + if remediate is not None: + content["remediate"] = remediate return JSONResponse( status_code=http_status.HTTP_422_UNPROCESSABLE_ENTITY, - content={"detail": rve.errors(), "body": body}, + content=content, ) diff --git a/roboco/api/websocket_bridge.py b/roboco/api/websocket_bridge.py index 998cdb0e..80654b02 100644 --- a/roboco/api/websocket_bridge.py +++ b/roboco/api/websocket_bridge.py @@ -22,7 +22,11 @@ async def _handle_notification_sent(event: Event) -> None: data = event.data notification_id_str = data.get("notification_id") - recipient_id_str = data.get("recipient_id") + # SENT events carry `recipient_id`; ACKED events carry `agent_id` (the + # agent who acknowledged). This handler serves both, so accept either — + # otherwise every acknowledgement logged a spurious "Incomplete + # notification event" and never reached the panel. + recipient_id_str = data.get("recipient_id") or data.get("agent_id") notification_type = data.get("type", "unknown") subject = data.get("subject", "") priority = data.get("priority", "normal") diff --git a/roboco/config.py b/roboco/config.py index 8c938027..e76a5ee1 100644 --- a/roboco/config.py +++ b/roboco/config.py @@ -323,7 +323,7 @@ class Settings(BaseSettings): description=( "Orchestrator-side directory where per-agent tool manifests are " "written. Must be a path that's bind-mounted from the host " - "(see docker-compose.yaml) so the docker daemon can in turn mount " + "(see docker-compose.yml) so the docker daemon can in turn mount " "the file into spawned agent containers as /app/tool-manifest.json." ), ) diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 12bb7c9e..81b8b540 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -3990,12 +3990,35 @@ Start now: evidence(task_id="{task_id}") await self._reap_with_service(svc) await db.commit() + def _assignee_has_active_instance(self, task: Any) -> bool: + """True if the task's assignee currently holds a live (ACTIVE) container. + + The heartbeat only approximates liveness. A developer deep in an + edit/test cycle can go longer than the heartbeat TTL between gateway + calls, so a heartbeat-only reaper releases claims out from under agents + that are alive and working — churning the task (and risking a double + spawn against the still-running container). The agent-instance registry + is the ground truth; defer to it when present. Defensive on missing + fields so a heartbeat-only caller (and the reaper's own unit tests) + behave exactly as before. + """ + owner = getattr(task, "assigned_to", None) or getattr(task, "claimed_by", None) + if not owner: + return False + instances = getattr(self, "_instances", None) + if not instances: + return False + instance = instances.get(self._resolve_agent_slug(str(owner))) + return instance is not None and instance.state == AgentState.ACTIVE + async def _reap_with_service(self, svc: "TaskService") -> None: """Inner reap loop, parameterized by the TaskService to use. Wraps each ``unclaim_for_reaper`` in try/except so a single bad row doesn't abort the dispatch tick — the reaper must keep ticking even - if one task's release somehow fails. + if one task's release somehow fails. A claim whose assignee still has + a live container is skipped: the heartbeat is a stale proxy there, and + reaping a working agent only churns the task. """ from roboco.utils.converters import require_uuid @@ -4004,6 +4027,8 @@ Start now: evidence(task_id="{task_id}") for t in candidates: ts = t.last_heartbeat_at if ts is None or ts < cutoff: + if self._assignee_has_active_instance(t): + continue task_id = require_uuid(t.id) try: await svc.unclaim_for_reaper(task_id) diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index ba2a235b..4a39f437 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -24,7 +24,6 @@ from roboco.services.gateway.choreographer._verb_runner import VerbRunner from roboco.services.gateway.claim_guards import ( already_active_guard, paused_tasks_guard, - sibling_sequence_guard, unmet_dependency_guard, ) from roboco.services.gateway.envelope import Envelope @@ -704,7 +703,6 @@ class Choreographer: *, agent_id: UUID, task: Any, - skip_sequence: bool = False, ) -> Envelope | None: """Run concurrency-invariant claim guards. Returns rejection or None. @@ -714,11 +712,7 @@ class Choreographer: in the verb's spec gate; the former role-typed and pm_cannot_execute_code guards have been deleted (Task 27, 2026-05-10). - Pre-gateway location: _helpers.py:124-204 + claim.py:121-180. - - ``skip_sequence`` lets resumption-of-already-claimed-task call sites - skip the sibling-sequence check (the sequence was already validated - on the original claim). + Pre-gateway location: _helpers.py:124-204. """ in_progress = await self.task.list_in_progress_for_agent(agent_id) if guard := already_active_guard(in_progress, task.id): @@ -730,26 +724,15 @@ class Choreographer: if dep_ids: unmet = await self.task.unmet_dependency_ids(dep_ids) if guard := unmet_dependency_guard(task, unmet): - return guard - if not skip_sequence: - siblings = await self._fetch_siblings(task) - if guard := sibling_sequence_guard(task, siblings): + # Park the dependency-gated task back to pending so the + # orchestrator stops respawning its assignee (the respawn loop + # targets only claimed/in_progress) and the dispatch dependency + # filter holds it until the upstream completes. No-op unless the + # task is currently claimed/in_progress. + await self.task.release_dependency_blocked_claim(task.id) return guard return None - async def _fetch_siblings(self, task: Any) -> list[Any]: - """Fetch sibling tasks for the sequence-order guard. - - Returns ``[]`` when the task has no parent (root task) so the - guard short-circuits. Otherwise returns the parent's subtasks via - ``TaskService.get_subtasks``. - """ - parent_id = getattr(task, "parent_task_id", None) - if parent_id is None: - return [] - siblings: list[Any] = await self.task.get_subtasks(parent_id) - return siblings - async def _non_terminal_subtask_ids(self, parent_task_id: UUID) -> str: """Return a human-readable comma-separated list of non-terminal subtasks. @@ -833,13 +816,11 @@ class Choreographer: task_id=task_id, verb=verb_name, ) - # Concurrency guards still apply (paused / already-active in another - # task). Sibling sequence is skipped on resumption — see - # _run_claim_guards docstring. + # Concurrency guards still apply on resumption (paused / already-active + # in another task). if guard := await self._run_claim_guards( agent_id=agent_id, task=t, - skip_sequence=True, ): return await self._emit_rejection( self._with_briefing(guard, briefing).with_introspection( @@ -899,7 +880,7 @@ class Choreographer: """Run all gates for an ``i_will_work_on`` / ``i_will_plan`` call. Order: spec.can_invoke_intent -> behavioral claim guards - (already_active / paused / sibling_sequence). Any rejection + (already_active / paused / unmet_dependency). Any rejection short-circuits with the appropriate envelope. Per-role claim authority (CLAIM_RULES) is enforced inside @@ -921,7 +902,7 @@ class Choreographer: # Behavioral pre-flight guards the spec doesn't yet model: # - already_active: agent has another in_progress task elsewhere # - paused_tasks: agent has a paused task they should resume first - # - sibling_sequence: an earlier-numbered sibling is still open + # - unmet_dependency: an upstream dependency is still non-terminal # The role/state/task_type checks already passed via the spec gate # above. These migrate into spec.extra_preconditions in a later # task; until then, keep them imperative so concurrency invariants @@ -1475,7 +1456,10 @@ class Choreographer: async def _i_am_done_gate(self, ctx: _IAmDoneContext) -> Envelope | None: """Run defense-in-depth tracing + field-level gates the spec doesn't model. - Returns the rejection envelope if any gate fails; None on pass. + Also pushes the branch to origin so a task cannot reach awaiting_qa + with commits that exist only in the developer's local workspace. + Returns the rejection envelope if any gate fails; None on pass. Shared + by the normal and resume-from-verifying paths so both push. """ if rejection := await self._check_tracing_gates( ctx.agent_id, ctx.task_id, ctx.task @@ -1485,12 +1469,38 @@ class Choreographer: ctx.agent_id, ctx.task_id, ctx.task ): return await self._reject_i_am_done(ctx, rejection) + if rejection := await self._ensure_branch_pushed(ctx): + return await self._reject_i_am_done(ctx, rejection) # Wave C5 (2026-05-12) — pre-gateway parity. Persist per-criterion # status now that all gates have passed. The write runs AFTER the # verdict so it cannot change i_am_done's rejection behavior. await self._write_criteria_status(ctx.agent_id, ctx.task_id, ctx.task) return None + async def _ensure_branch_pushed(self, ctx: _IAmDoneContext) -> Envelope | None: + """Push the task branch to origin before it reaches awaiting_qa. + + QA reviews the remote PR branch. A fix committed during a revision + cycle lives only in the developer's local workspace until pushed — + without this, QA re-reviews the stale remote and fails the same task + every cycle (a non-converging loop). Idempotent: a no-op when nothing + is unpushed, so first-submit (already pushed by open_pr) is unaffected. + """ + try: + await self.git.push_task_branch(ctx.agent_id, ctx.task_id) + except Exception as exc: + return Envelope.invalid_state( + message=f"could not push your branch to origin: {exc}", + remediate=( + "your latest commits are local-only and QA reviews the " + "pushed PR branch. resolve the push error (often a " + "transient network / fetch timeout) and call i_am_done " + "again." + ), + context_briefing=ctx.briefing, + ) + return None + @staticmethod def _extract_first_commit_sha(t: Any) -> str | None: """Read the first commit sha off the task, dict or model alike.""" @@ -3349,13 +3359,13 @@ class Choreographer: return value.value if hasattr(value, "value") else str(value) async def _wire_ux_frontend_dependency(self, new_task: Any, parent: Any) -> None: - """Cross-cell sequencing: in a product fan-out the FRONTEND cell task - depends on the UX/UI cell task — UX design is upstream of frontend - implementation, while backend runs in parallel. Wires the dependency in - either delegation order. A dev/code subtask delegated under a cell task - that is itself still waiting on that dependency inherits it, so the - developer is held until UX is done instead of coding ahead of the - design. Best-effort: never breaks delegate. + """Cross-cell sequencing: in a product fan-out the implementation cells + (FRONTEND and BACKEND) depend on the UX/UI cell task — UX design defines + the screens and API contracts both cells build against, so it is upstream + of implementation. Wires the dependency in either delegation order. A + dev/code subtask delegated under a cell task that is itself still waiting + on that dependency inherits it, so the developer is held until UX is done + instead of coding ahead of the design. Best-effort: never breaks delegate. """ if parent is None or getattr(parent, "product_id", None) is None: return @@ -3368,11 +3378,14 @@ class Choreographer: await self.task.inherit_unmet_dependencies(new_task.id, parent.id) if nt_team == Team.FRONTEND.value: await self._depend_frontend_on_ux(new_task, parent.id) + elif nt_team == Team.BACKEND.value: + await self._depend_backend_on_ux(new_task, parent.id) elif nt_team == Team.UX_UI.value: await self._depend_pending_frontends_on_ux(new_task, parent.id) + await self._depend_pending_backends_on_ux(new_task, parent.id) except Exception as exc: logger.warning( - "cross-cell UX->FE sequencing wiring failed", + "cross-cell UX->implementation sequencing wiring failed", error=str(exc), parent_task_id=str(getattr(parent, "id", None)), ) @@ -3396,6 +3409,32 @@ class Choreographer: ) if ux is not None: await self.task.add_dependency(fe_task.id, ux.id) + await self.task.set_sequence( + fe_task.id, (getattr(ux, "sequence", 0) or 0) + 1 + ) + + async def _depend_backend_on_ux(self, be_task: Any, parent_id: Any) -> None: + """Make a new BACKEND cell task wait on its non-terminal UX/UI sibling.""" + from roboco.foundation.identity import Team + from roboco.models.base import TaskStatus + + terminal = {TaskStatus.COMPLETED, TaskStatus.CANCELLED} + siblings = await self.task.get_subtasks(parent_id) + ux = next( + ( + s + for s in siblings + if self._team_value(s.team) == Team.UX_UI.value + and s.id != be_task.id + and s.status not in terminal + ), + None, + ) + if ux is not None: + await self.task.add_dependency(be_task.id, ux.id) + await self.task.set_sequence( + be_task.id, (getattr(ux, "sequence", 0) or 0) + 1 + ) async def _depend_pending_frontends_on_ux( self, ux_task: Any, parent_id: Any @@ -3405,6 +3444,7 @@ class Choreographer: from roboco.models.base import TaskStatus not_started = {TaskStatus.BACKLOG, TaskStatus.PENDING} + ux_sequence = (getattr(ux_task, "sequence", 0) or 0) + 1 siblings = await self.task.get_subtasks(parent_id) for fe in siblings: if ( @@ -3413,6 +3453,26 @@ class Choreographer: and fe.status in not_started ): await self.task.add_dependency(fe.id, ux_task.id) + await self.task.set_sequence(fe.id, ux_sequence) + + async def _depend_pending_backends_on_ux( + self, ux_task: Any, parent_id: Any + ) -> None: + """Retro-wire not-yet-started BACKEND siblings onto a new UX/UI task.""" + from roboco.foundation.identity import Team + from roboco.models.base import TaskStatus + + not_started = {TaskStatus.BACKLOG, TaskStatus.PENDING} + ux_sequence = (getattr(ux_task, "sequence", 0) or 0) + 1 + siblings = await self.task.get_subtasks(parent_id) + for be in siblings: + if ( + self._team_value(be.team) == Team.BACKEND.value + and be.id != ux_task.id + and be.status in not_started + ): + await self.task.add_dependency(be.id, ux_task.id) + await self.task.set_sequence(be.id, ux_sequence) async def _resolve_subtask_project( self, parent: Any, inputs: DelegateInputs @@ -3901,6 +3961,32 @@ class Choreographer: verb="unblock", ) + # A dependency block must not be cleared by hand. It auto-clears via + # _unblock_dependents the moment its last dependency reaches a terminal + # state; forcing it now would let the dependent proceed without the + # upstream's work (e.g. a frontend task built before its UX design lands). + dep_ids = list(t.dependency_ids or []) + unmet = await self.task.unmet_dependency_ids(dep_ids) if dep_ids else [] + if unmet: + return await self._emit_rejection( + Envelope.invalid_state( + message=( + f"task {task_id} still depends on {len(unmet)} " + "unfinished task(s); a dependency block clears on its " + "own once the upstream work completes" + ), + remediate=( + "don't force this — let the dependency finish; the task " + "auto-unblocks the moment its last dependency reaches " + "completed/cancelled" + ), + context_briefing=await self._briefing_for(pm_agent_id, task_id), + ).with_introspection(task=t, role=role), + agent_id=pm_agent_id, + task_id=task_id, + verb="unblock", + ) + if env := await self._check_pm_decision_required( "unblock", pm_agent_id, task_id, t ): diff --git a/roboco/services/gateway/choreographer/_protocol.py b/roboco/services/gateway/choreographer/_protocol.py index c085c853..a3091e73 100644 --- a/roboco/services/gateway/choreographer/_protocol.py +++ b/roboco/services/gateway/choreographer/_protocol.py @@ -72,7 +72,6 @@ class ChoreographerHelpers: *, agent_id: UUID, task: Any, - skip_sequence: bool = False, ) -> Envelope | None: raise NotImplementedError diff --git a/roboco/services/gateway/choreographer/doc.py b/roboco/services/gateway/choreographer/doc.py index afb48267..2923284b 100644 --- a/roboco/services/gateway/choreographer/doc.py +++ b/roboco/services/gateway/choreographer/doc.py @@ -178,7 +178,6 @@ class DocMixin(_Base): guard = await self._run_claim_guards( agent_id=doc_agent_id, task=t, - skip_sequence=True, ) if guard: guard.with_introspection(task=t, role=role_str) diff --git a/roboco/services/gateway/choreographer/qa.py b/roboco/services/gateway/choreographer/qa.py index dbe976e3..aad48acb 100644 --- a/roboco/services/gateway/choreographer/qa.py +++ b/roboco/services/gateway/choreographer/qa.py @@ -152,7 +152,6 @@ class QAMixin(_Base): guard = await self._run_claim_guards( agent_id=qa_agent_id, task=t, - skip_sequence=True, ) if guard: guard.with_introspection(task=t, role=role_str) diff --git a/roboco/services/gateway/claim_guards.py b/roboco/services/gateway/claim_guards.py index c36ca385..b372ab7f 100644 --- a/roboco/services/gateway/claim_guards.py +++ b/roboco/services/gateway/claim_guards.py @@ -13,7 +13,6 @@ through ``spec.can_invoke_action``'s CLAIM_RULES + ``ActionSpec Pre-gateway location at commit 0c3d15a: roboco/mcp/tasks/handlers/_helpers.py:124-204 - roboco/mcp/tasks/handlers/claim.py:121-180 (sibling sequence) """ from __future__ import annotations @@ -31,10 +30,6 @@ _ACTIVE_BLOCKING_STATUSES: frozenset[str] = frozenset( {"claimed", "in_progress", "verifying"} ) -# Terminal statuses that satisfy the sibling-sequence check — -# pre-gateway claim.py:153. -_TERMINAL_STATUSES: frozenset[str] = frozenset({"completed", "cancelled"}) - def already_active_guard( in_progress_tasks: list[Any], target_task_id: UUID @@ -107,49 +102,3 @@ def unmet_dependency_guard( "completed/cancelled before claiming this task" ), ) - - -def _earlier_blocking_sibling( - target_task: Any, siblings: list[Any], my_sequence: int -) -> Any | None: - """Return the first non-terminal sibling with a lower sequence, else None.""" - for sib in siblings: - if sib.id == target_task.id: - continue - sib_seq = getattr(sib, "sequence", 0) or 0 - sib_status = str(getattr(sib, "status", "")) - if sib_seq < my_sequence and sib_status not in _TERMINAL_STATUSES: - return sib - return None - - -def sibling_sequence_guard(target_task: Any, siblings: list[Any]) -> Envelope | None: - """Refuse claim if any earlier-sequence sibling is non-terminal. - - Pre-gateway: claim.py:_validate_sibling_sequence 121-180. - - A task with sequence=N is blocked while any sibling with sequence int: + """Idempotently push a task's branch to origin; return commits pushed. + + Reviewers see the remote PR branch, not the developer's workspace. A + fix committed during a revision cycle lives only in that local clone + until it is pushed — so without an explicit push at the QA-submission + boundary, QA re-reviews the stale remote and fails the same task on + every cycle. Self-resolves the project/workspace from the task so the + choreographer can call it with just (agent, task). A no-op when there + is nothing unpushed; raises typed service errors on a real failure. + """ + task = await self._assert_task_owned_with_branch(task_id, agent_id) + project = await self._project_for_task(task) + if project is None: + return 0 + workspace = await self.get_workspace(project.slug, agent_id) + await self._assert_on_task_branch(workspace, task.branch_name) + _branch, pushed = await self.push(workspace) + return pushed + # ========================================================================= # PR METHODS # ========================================================================= diff --git a/roboco/services/task.py b/roboco/services/task.py index 2e7eac38..103168e1 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -407,22 +407,44 @@ class TaskService(BaseService): error=str(e), ) - # Fire-and-forget audit write. Critical: we must hold a strong - # reference to the Task object (via `_background_tasks`) — the event - # loop only weak-refs tasks, so without this the audit write can be - # garbage-collected before it commits. That's why audit_log was - # coming up empty even though the log call ran. + self._emit_status_transition_audit( + task, + from_status=current, + to_status=target, + agent_role=agent_role, + audit_agent_id=audit_agent_id, + ) + + def _emit_status_transition_audit( + self, + task: TaskTable, + *, + from_status: str, + to_status: str, + agent_role: str | None, + audit_agent_id: str | UUID | None, + ) -> None: + """Emit the ``task.`` audit row for a status transition. + + Extracted from ``_validate_and_set_status`` so transition paths that + set ``task.status`` directly — e.g. ``apply_escalation``, which blocks a + task without routing through the strict transition validator — record + the same audit event. No status change may bypass the audit log. + + Fire-and-forget, but we hold a strong reference to the background task + (via ``_background_tasks``): the event loop only weak-refs tasks, so + without it the audit write can be garbage-collected before it commits. + + The explicit ``audit_agent_id`` (capture-before-mutate) wins: callers + like ``submit_for_qa`` clear ``task.claimed_by`` before transitioning + but still want the row attributed to the outgoing agent. Otherwise fall + back to ``task.claimed_by``. + """ import asyncio import contextlib from roboco.services.audit import get_audit_service - # Prefer the explicit `audit_agent_id` when the caller passed one - # (capture-before-mutate pattern: callers like `submit_for_qa` and - # `pass_qa` clear `task.claimed_by` BEFORE calling us so the next - # role can claim, but still want the audit row attributed to the - # outgoing agent). Fall back to `task.claimed_by` for transitions - # where the assignment didn't change (claim, start_work, etc.). if audit_agent_id is not None: resolved_audit_agent_id: str | None = str(audit_agent_id) elif task.claimed_by is not None: @@ -434,12 +456,12 @@ class TaskService(BaseService): with contextlib.suppress(RuntimeError): bg = asyncio.get_running_loop().create_task( audit.log_task_event( - event_type=f"task.{target}", + event_type=f"task.{to_status}", task_id=str(task.id), agent_id=resolved_audit_agent_id, details={ - "from_status": current, - "to_status": target, + "from_status": from_status, + "to_status": to_status, "agent_role": agent_role, "team": ( task.team.value @@ -1697,13 +1719,33 @@ class TaskService(BaseService): error=str(e), ) + @staticmethod + def _resolve_doc_abspath(rel_path: str) -> str: + """Resolve a documenter-supplied doc path to its on-disk absolute path. + + Docs live under ``DOCS_BASE_PATH`` (``/app/docs``). Agents sometimes + hand a path already rooted at ``docs/`` (or an absolute path); joining + ``DOCS_BASE_PATH`` with a ``docs/``-prefixed relative path doubles the + segment (``/app/docs/docs/...``), so the file is never found and the + docs never index into RAG. Normalize: trust an absolute path; otherwise + strip a single redundant leading ``docs/`` before joining. + """ + from pathlib import Path + + from roboco.services.docs import DOCS_BASE_PATH + + path = Path(rel_path) + if path.is_absolute(): + return str(path) + parts = path.parts + if parts and parts[0] == DOCS_BASE_PATH.name: + path = Path(*parts[1:]) if len(parts) > 1 else Path() + return str(DOCS_BASE_PATH / path) + async def _index_docs_background( self, task_id: UUID, documents: list[dict[str, Any]] ) -> None: """Index documentation from completed doc task (fire-and-forget).""" - from pathlib import Path - - from roboco.services.docs import DOCS_BASE_PATH from roboco.services.optimal import get_optimal_service try: @@ -1714,8 +1756,7 @@ class TaskService(BaseService): for d in documents: rel_path = d.get("path") if rel_path: - absolute_path = str(DOCS_BASE_PATH / Path(rel_path)) - doc_paths.append(absolute_path) + doc_paths.append(self._resolve_doc_abspath(rel_path)) if doc_paths: count = await optimal.index_documentation(doc_paths, project="roboco") @@ -2074,24 +2115,65 @@ class TaskService(BaseService): ownership/role checks because the holder is provably dead (no heartbeat past TTL). """ + await self._force_unclaim_to_pending(task_id, reason="reaper-unclaim") + + async def release_dependency_blocked_claim(self, task_id: UUID) -> None: + """Release a claimed/in_progress task whose dependency is still unmet. + + A task assigned with an unfinished dependency cannot proceed, but while + it sits claimed/in_progress the orchestrator keeps respawning its + assignee (the respawn loop targets only claimed/in_progress). Releasing + it to pending stops that churn: the dispatch dependency filter holds it + un-spawned, and ``_unblock_dependents`` clears the dependency once the + upstream completes so it re-dispatches on its own. ``claimed -> blocked`` + is not a legal transition, so pending (held by the dependency filter) is + the lifecycle-correct resting state. No-op when not in a releasable state. + + Also forgets ``branch_name`` so the eventual re-claim re-runs branch + creation and cuts the branch fresh off the current integration tip — + which by then includes the upstream's merged work — instead of reusing a + snapshot taken before the dependency landed. A dependency-blocked task + has done no work of its own, so nothing is lost; ``create_branch`` + leaves any branch carrying real commits intact. + """ + if not await self._force_unclaim_to_pending(task_id, reason="dependency-unmet"): + return + task = await self.get(task_id) + if task is not None and task.branch_name: + task.branch_name = None + await self.session.flush() + + async def _force_unclaim_to_pending(self, task_id: UUID, *, reason: str) -> bool: + """Force a claimed/in_progress task back to pending (system action). + + Shared core of ``unclaim_for_reaper`` and + ``release_dependency_blocked_claim``. Routes through + ``_validate_and_set_status`` so the state machine records the + transition, clears assignee/heartbeat/claimant, and abandons the active + WorkSession (best-effort, tagged with ``reason``) so a re-claim doesn't + trip the uniqueness constraint. Bypasses ownership/role checks — the + system itself is performing the transition. Returns True iff the task + was actually released (False when missing or not in a releasable state). + """ task = await self.get(task_id) if task is None: - return + return False if task.status not in (TaskStatus.CLAIMED, TaskStatus.IN_PROGRESS): - return + return False try: self._validate_and_set_status(task, TaskStatus.PENDING, None) except TaskLifecycleError: - return + return False if task.work_session_id: await self._abandon_work_session_best_effort( - task.work_session_id, reason="reaper-unclaim" + task.work_session_id, reason=reason ) task.work_session_id = cast("Any", None) task.assigned_to = cast("Any", None) task.last_heartbeat_at = None task.active_claimant_id = cast("Any", None) await self.session.flush() + return True async def _abandon_work_session_best_effort( self, session_id: Any, *, reason: str @@ -3346,6 +3428,15 @@ class TaskService(BaseService): return if task.assigned_to and not task.blocker_raised_by: task.blocker_raised_by = cast("Any", task.assigned_to) + # Capture before mutating: the audit row must record the real prior + # status and attribute the block to the outgoing owner, not the + # escalation target we are about to assign. + pre_block_status = ( + task.status.value + if isinstance(task.status, TaskStatus) + else str(task.status) + ) + pre_block_owner = cast("Any", task.claimed_by) task.assigned_to = cast("Any", target_agent_id) task.claimed_by = cast("Any", target_agent_id) task.status = TaskStatus.BLOCKED @@ -3355,6 +3446,16 @@ class TaskService(BaseService): ) task.dev_notes = existing_notes + escalation_note await self.session.flush() + # This path sets BLOCKED directly (bypassing the strict transition + # validator), so emit the task.blocked audit explicitly — no status + # change may skip the audit log. + self._emit_status_transition_audit( + task, + from_status=pre_block_status, + to_status=TaskStatus.BLOCKED.value, + agent_role=None, + audit_agent_id=pre_block_owner, + ) self.log.info( "Task escalated and blocked", task_id=str(task.id), @@ -4253,6 +4354,23 @@ class TaskService(BaseService): task.dependency_ids = [*task.dependency_ids, depends_on_id] await self.session.flush() + async def set_sequence(self, task_id: UUID, sequence: int) -> None: + """Set a task's sibling-ordering sequence (lower = first). + + `sequence` is a display / dispatch-priority field only — it orders + siblings in `list_pending`, `list_for_team`, and the panel and carries + no claim-gating semantics (dependencies gate claims). Cross-cell + fan-out uses it so an upstream design task sorts ahead of the + implementation tasks that depend on it. No-op if the task is gone or + already at `sequence`. + """ + task = await self.get(task_id) + if task is None: + return + if task.sequence != sequence: + task.sequence = sequence + await self.session.flush() + async def unmet_dependency_ids(self, dependency_ids: list[UUID]) -> list[UUID]: """Return the subset of dependency IDs whose status is non-terminal. diff --git a/roboco/services/workspace.py b/roboco/services/workspace.py index f6f84752..1a5757ed 100644 --- a/roboco/services/workspace.py +++ b/roboco/services/workspace.py @@ -422,11 +422,14 @@ class WorkspaceService: Called from `ensure_workspace`'s healthy short-circuit so that a respawned PM/Doc reads fresh `origin/` refs instead of - whatever the previous spawn left on disk. We deliberately omit a - positional refspec — `git fetch origin` (no args after `origin`) - updates every branch under `refs/remotes/origin/`, which is what - downstream `git diff origin/` and `git log origin/` - readers want. + whatever the previous spawn left on disk. The fetch is SCOPED to the + workspace's current branch + the repo's default branch (with + `--no-tags --prune`). An all-refs `git fetch origin` transfers every + accumulated `feature/*` on a monorepo and blows past the timeout, after + which the workspace silently keeps a stale base and the agent builds on + it. The refs a workspace's `git diff/log origin/` readers need + are its own branch and the default; the integration branch is refreshed + at branch-creation time (`create_branch_for_task`), not here. No `-c http.extraheader=…` token injection: the orchestrator did the original clone with a token but `_configure_git()` already @@ -442,9 +445,31 @@ class WorkspaceService: remote is operationally bad. """ + def _git(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["git", *args], + cwd=str(workspace), + capture_output=True, + text=True, + check=False, + ) + + def _scoped_refs() -> list[str]: + """The current branch + the repo's default branch, deduped.""" + current = _git("rev-parse", "--abbrev-ref", "HEAD").stdout.strip() + origin_head = _git( + "symbolic-ref", "--short", "refs/remotes/origin/HEAD" + ).stdout.strip() + default = origin_head.split("/", 1)[1] if "/" in origin_head else "master" + refs: list[str] = [] + for ref in (current, default): + if ref and ref != "HEAD" and ref not in refs: + refs.append(ref) + return refs or ["master"] + def _do_fetch() -> subprocess.CompletedProcess[str]: return subprocess.run( - ["git", "fetch", "origin"], + ["git", "fetch", "--no-tags", "--prune", "origin", *_scoped_refs()], cwd=str(workspace), capture_output=True, text=True, diff --git a/tests/foundation/test_lifecycle_consumer_parity.py b/tests/foundation/test_lifecycle_consumer_parity.py index 3ce35905..6e484599 100644 --- a/tests/foundation/test_lifecycle_consumer_parity.py +++ b/tests/foundation/test_lifecycle_consumer_parity.py @@ -1231,8 +1231,8 @@ async def test_claim_review_matches_spec(role: str, status: str) -> None: The verb body owns dispatch via ``task.qa_claim`` (not the runner's claim+start chain) because the runtime semantic is "QA inspects, status stays at awaiting_qa" — see qa.py module docstring. The - behavioral claim guards (already_active / paused / sibling_sequence - skipped) run after the spec gate; they're not modelled by the spec. + behavioral claim guards (already_active / paused / unmet_dependency) + run after the spec gate; they're not modelled by the spec. """ agent_id = uuid4() task_id = uuid4() diff --git a/tests/integration/test_dev_subtask_inherits_cross_cell_dependency.py b/tests/integration/test_dev_subtask_inherits_cross_cell_dependency.py index ad6acf22..e590a109 100644 --- a/tests/integration/test_dev_subtask_inherits_cross_cell_dependency.py +++ b/tests/integration/test_dev_subtask_inherits_cross_cell_dependency.py @@ -77,13 +77,21 @@ async def fanout_setup(db_session: AsyncSession) -> AsyncIterator[dict]: assigned_cell=Team.UX_UI, created_by=system.id, ) + be_project = ProjectTable( + id=uuid4(), + name="BE", + slug=f"be-{uuid4().hex[:6]}", + git_url="https://example.com/be.git", + assigned_cell=Team.BACKEND, + created_by=system.id, + ) product = ProductTable( id=uuid4(), name="Prod", slug=f"prod-{uuid4().hex[:6]}", created_by=system.id, ) - db_session.add_all([fe_project, ux_project, product]) + db_session.add_all([fe_project, ux_project, be_project, product]) await db_session.flush() svc = TaskService(db_session) @@ -105,6 +113,7 @@ async def fanout_setup(db_session: AsyncSession) -> AsyncIterator[dict]: "fe_dev_id": fe_dev.id, "fe_project_id": fe_project.id, "ux_project_id": ux_project.id, + "be_project_id": be_project.id, "product_id": product.id, } @@ -224,3 +233,145 @@ async def test_dev_subtask_held_until_ux_dependency_resolves( assert dev_subtask.id in pending_after_ids, ( "dev subtask must become dispatchable once UX reaches a terminal state" ) + + +@pytest.mark.asyncio +async def test_dependent_cell_sequence_follows_upstream_ux( + fanout_setup: dict, +) -> None: + """The frontend cell task sorts after its UX upstream: wiring the + dependency also bumps its sequence to the UX task's sequence + 1, so + list ordering and the panel show UX before the work it gates.""" + svc: TaskService = fanout_setup["svc"] + tree = await _build_product_fanout(fanout_setup) + ux_row = await svc.get(tree["ux_cell"].id) + fe_row = await svc.get(tree["fe_cell"].id) + assert ux_row is not None and fe_row is not None + assert fe_row.sequence == (ux_row.sequence or 0) + 1, ( + "the dependent frontend task must sort one step after its UX upstream" + ) + + +@pytest.mark.asyncio +async def test_backend_cell_also_depends_on_ux(fanout_setup: dict) -> None: + """UX/UI design defines the API contracts the backend builds against, so a + backend cell task in the same fan-out also waits on the UX cell task and + sorts after it.""" + svc: TaskService = fanout_setup["svc"] + choreo: Choreographer = fanout_setup["choreo"] + tree = await _build_product_fanout(fanout_setup) + root = tree["root"] + ux_cell = tree["ux_cell"] + + be_cell = await svc.create_subtask( + TaskCreateRequest( + title="Backend implementation for the feature", + description="a real backend cell task description over twenty chars", + acceptance_criteria=["endpoints satisfy the contract"], + team=Team.BACKEND, + created_by=fanout_setup["creator"], + project_id=fanout_setup["be_project_id"], + product_id=fanout_setup["product_id"], + parent_task_id=root.id, + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.MEDIUM, + ) + ) + # Forward order: the backend cell is delegated after the UX cell exists. + await choreo._wire_ux_frontend_dependency(be_cell, root) + await svc.session.flush() + + be_row = await svc.get(be_cell.id) + ux_row = await svc.get(ux_cell.id) + assert be_row is not None and ux_row is not None + assert ux_cell.id in be_row.dependency_ids, ( + "backend cell task must depend on the UX cell task" + ) + assert be_row.sequence == (ux_row.sequence or 0) + 1, ( + "the backend task must sort one step after its UX upstream" + ) + + +@pytest.mark.asyncio +async def test_pending_impl_cells_retrowired_when_ux_arrives_later( + fanout_setup: dict, +) -> None: + """When the UX cell task is delegated AFTER still-pending frontend and + backend siblings, both are retro-wired onto UX and sorted after it — the + 'either delegation order' guarantee, for both implementation cells.""" + svc: TaskService = fanout_setup["svc"] + choreo: Choreographer = fanout_setup["choreo"] + + root = await svc.create( + TaskCreateRequest( + title="Build the feature (board fan-out)", + description="a real coordination task description over twenty chars", + acceptance_criteria=["delegated to frontend + backend + ux_ui cells"], + team=Team.BOARD, + created_by=fanout_setup["creator"], + project_id=None, + product_id=fanout_setup["product_id"], + task_type=TaskType.CODE, + nature=TaskNature.NON_TECHNICAL, + estimated_complexity=Complexity.HIGH, + ) + ) + fe_cell = await svc.create_subtask( + TaskCreateRequest( + title="Frontend implementation for the feature", + description="a real frontend cell task description over twenty chars", + acceptance_criteria=["UI matches the design"], + team=Team.FRONTEND, + created_by=fanout_setup["creator"], + project_id=fanout_setup["fe_project_id"], + product_id=fanout_setup["product_id"], + parent_task_id=root.id, + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.MEDIUM, + ) + ) + be_cell = await svc.create_subtask( + TaskCreateRequest( + title="Backend implementation for the feature", + description="a real backend cell task description over twenty chars", + acceptance_criteria=["endpoints satisfy the contract"], + team=Team.BACKEND, + created_by=fanout_setup["creator"], + project_id=fanout_setup["be_project_id"], + product_id=fanout_setup["product_id"], + parent_task_id=root.id, + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.MEDIUM, + ) + ) + # UX is delegated LAST — both pending implementation cells must be wired. + ux_cell = await svc.create_subtask( + TaskCreateRequest( + title="UX/UI design for the feature", + description="a real ux design task description over twenty chars", + acceptance_criteria=["wireframes approved"], + team=Team.UX_UI, + created_by=fanout_setup["creator"], + project_id=fanout_setup["ux_project_id"], + product_id=fanout_setup["product_id"], + parent_task_id=root.id, + task_type=TaskType.DESIGN, + nature=TaskNature.TECHNICAL, + estimated_complexity=Complexity.MEDIUM, + ) + ) + await choreo._wire_ux_frontend_dependency(ux_cell, root) + await svc.session.flush() + + fe_row = await svc.get(fe_cell.id) + be_row = await svc.get(be_cell.id) + ux_row = await svc.get(ux_cell.id) + assert fe_row is not None and be_row is not None and ux_row is not None + assert ux_cell.id in fe_row.dependency_ids, "frontend must retro-wire onto UX" + assert ux_cell.id in be_row.dependency_ids, "backend must retro-wire onto UX" + expected_sequence = (ux_row.sequence or 0) + 1 + assert fe_row.sequence == expected_sequence + assert be_row.sequence == expected_sequence diff --git a/tests/integration/test_full_lifecycle_real_db.py b/tests/integration/test_full_lifecycle_real_db.py index 8829c265..83bb956a 100644 --- a/tests/integration/test_full_lifecycle_real_db.py +++ b/tests/integration/test_full_lifecycle_real_db.py @@ -120,6 +120,10 @@ class _StubGit: del branch_name, actor_agent_id return ("ok", 0) + async def push_task_branch(self, agent_id: UUID, task_id: UUID) -> int: + del agent_id, task_id + return 0 + async def create_pr( self, branch_name: str, diff --git a/tests/integration/test_lifecycle_real_db.py b/tests/integration/test_lifecycle_real_db.py index fcd5d3d7..b2731e20 100644 --- a/tests/integration/test_lifecycle_real_db.py +++ b/tests/integration/test_lifecycle_real_db.py @@ -124,6 +124,10 @@ class _StubGit: del branch_name, actor_agent_id return ("ok", 0) + async def push_task_branch(self, agent_id: UUID, task_id: UUID) -> int: + del agent_id, task_id + return 0 + async def create_pr( self, branch_name: str, diff --git a/tests/integration/test_pre_assigned_dev_held_on_unmet_dependency.py b/tests/integration/test_pre_assigned_dev_held_on_unmet_dependency.py index 578b062b..27da688b 100644 --- a/tests/integration/test_pre_assigned_dev_held_on_unmet_dependency.py +++ b/tests/integration/test_pre_assigned_dev_held_on_unmet_dependency.py @@ -322,3 +322,46 @@ async def test_all_three_dev_paths_gate_then_release(dep_gate_setup: dict) -> No assert ( await choreo._run_claim_guards(agent_id=fe_dev_db_id, task=released) is None ), "claim guard must allow once UX is terminal" + + +@pytest.mark.asyncio +async def test_claimed_dependency_blocked_task_is_released_to_pending( + dep_gate_setup: dict, +) -> None: + """A CLAIMED task whose dependency is unmet is released back to pending. + + Unlike the pre-assigned-but-pending dev subtask, a cell task can reach + CLAIMED with an unfinished dependency (the PM claims it before the upstream + resolves). Left claimed, the orchestrator's respawn loop churns its + assignee. The claim guard now releases it to pending — ``claimed -> blocked`` + is not a legal transition, so pending (held by the dependency filter) is the + lifecycle-correct resting state, and ``_unblock_dependents`` re-dispatches it + once the upstream completes. + """ + svc: TaskService = dep_gate_setup["svc"] + choreo: Choreographer = dep_gate_setup["choreo"] + fe_dev_db_id = dep_gate_setup["fe_dev_db_id"] + + tree = await _seed_dev_subtask_with_unmet_dep(dep_gate_setup) + dev_subtask = tree["dev_subtask"] + + # Force the held task to CLAIMED (the state a respawn loop churns on). + dev_subtask.status = TaskStatus.CLAIMED + dev_subtask.branch_name = "feature/frontend/DEVLEAF01" + await svc.session.flush() + + held = await svc.get(dev_subtask.id) + guard = await choreo._run_claim_guards(agent_id=fe_dev_db_id, task=held) + assert guard is not None, "claim guard must still reject while UX is unmet" + assert guard.error == "invalid_state" + + after = await svc.get(dev_subtask.id) + assert after is not None + assert after.status == TaskStatus.PENDING, ( + "a claimed dependency-blocked task must be released to pending" + ) + assert after.assigned_to is None, "release clears the assignee" + assert after.branch_name is None, ( + "release clears branch_name so the re-claim cuts fresh off the current " + "integration tip (which by then includes the upstream's work)" + ) diff --git a/tests/unit/api/test_app.py b/tests/unit/api/test_app.py index fe55929a..26b1f19c 100644 --- a/tests/unit/api/test_app.py +++ b/tests/unit/api/test_app.py @@ -8,7 +8,7 @@ extraction, optimal-service). from __future__ import annotations -from contextlib import asynccontextmanager +from contextlib import ExitStack, asynccontextmanager from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -141,3 +141,61 @@ async def test_lifespan_handles_optimal_init_failure_gracefully() -> None: app = create_app() async with lifespan(app): assert app.state.optimal is None + + +def _lifespan_io_patches() -> list: + transcription_mock = MagicMock() + transcription_mock.start = AsyncMock() + transcription_mock.stop = AsyncMock() + return [ + patch("roboco.api.app.init_db", new=AsyncMock()), + patch("roboco.api.app.close_db", new=AsyncMock()), + patch("roboco.api.app.TranscriptionService", return_value=transcription_mock), + patch("roboco.api.app.ExtractionService"), + patch("roboco.api.app.ExtractionPipeline"), + patch( + "roboco.api.app.get_optimal_service", + new=AsyncMock(return_value=MagicMock()), + ), + patch("roboco.api.app.close_optimal_service", new=AsyncMock()), + ] + + +def _header_trust_warnings(logger_mock: MagicMock) -> list: + return [ + c + for c in logger_mock.warning.call_args_list + if c.args and "HEADER-TRUST" in c.args[0] + ] + + +async def _run_lifespan_with(*, auth_required: bool, logger_mock: MagicMock) -> None: + """Run the lifespan with heavy I/O patched and the auth flag forced.""" + with ExitStack() as stack: + for cm in _lifespan_io_patches(): + stack.enter_context(cm) + stack.enter_context( + patch("roboco.api.app._auth_required", return_value=auth_required) + ) + stack.enter_context(patch("roboco.api.app.logger", logger_mock)) + app = create_app() + async with lifespan(app): + pass + + +@pytest.mark.asyncio +async def test_lifespan_warns_in_header_trust_mode() -> None: + """Startup warns when agent auth is not enforced (header-trust mode).""" + logger_mock = MagicMock() + await _run_lifespan_with(auth_required=False, logger_mock=logger_mock) + assert _header_trust_warnings(logger_mock), ( + "header-trust startup warning expected when auth is not required" + ) + + +@pytest.mark.asyncio +async def test_lifespan_no_header_trust_warning_when_auth_required() -> None: + """No header-trust warning when ROBOCO_AGENT_AUTH_REQUIRED enforces tokens.""" + logger_mock = MagicMock() + await _run_lifespan_with(auth_required=True, logger_mock=logger_mock) + assert not _header_trust_warnings(logger_mock) diff --git a/tests/unit/api/test_middleware.py b/tests/unit/api/test_middleware.py index 059b67f1..c0d4b0c7 100644 --- a/tests/unit/api/test_middleware.py +++ b/tests/unit/api/test_middleware.py @@ -4,10 +4,16 @@ from __future__ import annotations from http import HTTPStatus +# UUID annotates a Pydantic model field below, so it must stay a runtime import +# (Pydantic resolves the annotation when building the model) despite `from +# __future__ import annotations` making it look type-checking-only to ruff. +from uuid import UUID # noqa: TC003 + from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient from pydantic import BaseModel from roboco.api.middleware import ( + _uuid_field_remediation, get_status_code, setup_middleware, ) @@ -193,6 +199,62 @@ def test_generic_exception_returns_500() -> None: assert "error" in body +# --------------------------------------------------------------------------- +# _uuid_field_remediation + truncated-task_id 422 remediation +# --------------------------------------------------------------------------- + + +def test_uuid_field_remediation_hits_truncated_task_id() -> None: + errors = [{"loc": ("body", "task_id"), "type": "uuid_parsing", "msg": "bad"}] + hint = _uuid_field_remediation(errors) + assert hint is not None + assert "full" in hint.lower() + assert "uuid" in hint.lower() + + +def test_uuid_field_remediation_ignores_other_field_errors() -> None: + errors = [{"loc": ("body", "title"), "type": "string_too_short", "msg": "x"}] + assert _uuid_field_remediation(errors) is None + + +def test_uuid_field_remediation_ignores_non_uuid_task_id_errors() -> None: + errors = [{"loc": ("body", "task_id"), "type": "missing", "msg": "required"}] + assert _uuid_field_remediation(errors) is None + + +class _TaskIdBody(BaseModel): + task_id: UUID + + +def _make_uuid_app() -> FastAPI: + app = FastAPI() + + @app.post("/needs-uuid") + async def _need(body: _TaskIdBody) -> dict: + return {"task_id": str(body.task_id)} + + setup_middleware(app) + return app + + +def test_truncated_task_id_422_carries_remediation() -> None: + """An 8-char task_id (the recurring agent mistake) returns 422 + remediate.""" + client = TestClient(_make_uuid_app(), raise_server_exceptions=False) + response = client.post("/needs-uuid", json={"task_id": "cee99ecc"}) + assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + body = response.json() + assert "remediate" in body + assert "full" in body["remediate"].lower() + + +def test_other_validation_422_omits_remediation() -> None: + """A non-task_id validation error keeps the standard 422 shape (no remediate).""" + client = TestClient(_make_uuid_app(), raise_server_exceptions=False) + response = client.post("/needs-uuid", json={}) # missing task_id entirely + assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + assert "remediate" not in response.json() + + def test_request_validation_handler_returns_422_with_details() -> None: """request_validation_handler logs + returns 422 with errors+body (251-260).""" diff --git a/tests/unit/api/test_websocket_bridge.py b/tests/unit/api/test_websocket_bridge.py index 990e05ed..48e12c39 100644 --- a/tests/unit/api/test_websocket_bridge.py +++ b/tests/unit/api/test_websocket_bridge.py @@ -104,6 +104,30 @@ async def test_handle_notification_sent_broadcasts_when_connected() -> None: assert call_kwargs["agent_ids"] == [rid] +@pytest.mark.asyncio +async def test_handle_notification_acked_broadcasts_using_agent_id() -> None: + """ACKED events carry `agent_id`, not `recipient_id`; the shared handler + must still forward (to the acking agent) rather than log 'Incomplete + notification event' on every acknowledgement.""" + nid = uuid4() + aid = uuid4() + event = _evt( + EventType.NOTIFICATION_ACKED, + {"notification_id": str(nid), "agent_id": str(aid), "ack_type": "read"}, + ) + bcast = AsyncMock() + with ( + patch("roboco.api.websocket_bridge.broadcast_notification", bcast), + patch("roboco.api.websocket_bridge.manager") as mgr, + ): + mgr.notification_connections = {aid: {"socket-1"}} + await _handle_notification_sent(event) + bcast.assert_awaited_once() + call_kwargs = bcast.await_args.kwargs + assert call_kwargs["notification_id"] == nid + assert call_kwargs["agent_ids"] == [aid] + + # --------------------------------------------------------------------------- # _handle_session_event # --------------------------------------------------------------------------- diff --git a/tests/unit/gateway/test_choreographer_claim_guards.py b/tests/unit/gateway/test_choreographer_claim_guards.py index a905e0f0..a91893e9 100644 --- a/tests/unit/gateway/test_choreographer_claim_guards.py +++ b/tests/unit/gateway/test_choreographer_claim_guards.py @@ -1,15 +1,13 @@ """Gate Set A: claim-time guards restored from pre-gateway _helpers.py:124-204. Predicates ported into Choreographer claim verbs: -- SEQUENCE_ORDER_VIOLATION (earlier sibling must be terminal) - ALREADY_ACTIVE (no claim while in_progress task is open) - PAUSED_TASKS_EXIST (no claim while paused tasks exist) - PM_CANNOT_EXECUTE_CODE (cell_pm/main_pm cannot claim task_type=code) - ROLE_TYPED_CLAIM (developer/qa/documenter cannot cross-claim) These mirror pre-gateway gates at commit 0c3d15a, file -roboco/mcp/tasks/handlers/_helpers.py lines 124-204 plus -roboco/mcp/tasks/handlers/claim.py:121-180 for the sibling sequence check. +roboco/mcp/tasks/handlers/_helpers.py lines 124-204. """ from __future__ import annotations @@ -33,21 +31,6 @@ _STEPS = [ ), } ] -# Full parity: a fresh dev claim authors the same rich plan a PM does. -# These satisfy _dev_plan_gate (plan/approach >= 150 chars, -# technical_considerations, risks). -_GOOD_PLAN = ( - "Append the timestamp HTML comment to the very bottom of README.md without " - "touching any other line, then commit it on the task branch and open a PR. " - "Verify the diff is a single-line addition before submitting for QA." -) -_GOOD_TC = ["Use a trailing newline so the comment sits on its own line."] -_GOOD_RISKS = [ - { - "risk": "An accidental reformat of README.md balloons the diff.", - "mitigation": "Append only; assert the diff touches one line pre-commit.", - } -] def _make_deps(**overrides: Any) -> ChoreographerDeps: @@ -122,138 +105,6 @@ def _task_svc_with( return task_svc -# --------------------------------------------------------------------------- -# A.1 SEQUENCE_ORDER_VIOLATION -# --------------------------------------------------------------------------- - - -@pytest.mark.asyncio -async def test_i_will_work_on_blocks_when_earlier_sibling_open() -> None: - """Sequence=2 cannot be claimed while sequence=1 sibling is still open.""" - agent_id = uuid4() - parent_id = uuid4() - target_id = uuid4() - earlier_id = uuid4() - target = MagicMock( - id=target_id, - status="pending", - plan=None, - assigned_to=None, - parent_task_id=parent_id, - sequence=2, - task_type="code", - team="backend", - ) - earlier = MagicMock( - id=earlier_id, - status="in_progress", - sequence=1, - title="Earlier sibling", - ) - later = MagicMock( - id=target_id, - status="pending", - sequence=2, - ) - task_svc = _task_svc_with(target, lookups={"siblings": [earlier, later]}) - deps = _make_deps(task=task_svc) - c = Choreographer(deps) - - env = await c.i_will_work_on(agent_id, target_id, plan="x", steps=_STEPS) - body = env.as_dict() - assert body["error"] == "invalid_state" - assert "sequence" in body["message"].lower() - assert str(earlier_id) in body["remediate"] - task_svc.claim.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_i_will_work_on_allows_when_earlier_sibling_terminal() -> None: - """Earlier siblings completed/cancelled do not block.""" - agent_id = uuid4() - parent_id = uuid4() - target_id = uuid4() - target = MagicMock( - id=target_id, - status="pending", - plan={"x": 1}, - assigned_to=None, - parent_task_id=parent_id, - sequence=2, - task_type="code", - team="backend", - ) - earlier_done = MagicMock(id=uuid4(), status="completed", sequence=1) - earlier_cancelled = MagicMock(id=uuid4(), status="cancelled", sequence=0) - self_row = MagicMock(id=target_id, status="pending", sequence=2) - task_svc = _task_svc_with( - target, - agent_id=agent_id, - lookups={"siblings": [earlier_done, earlier_cancelled, self_row]}, - ) - task_svc.claim.return_value = MagicMock( - id=target_id, - status="claimed", - plan={"x": 1}, - assigned_to=agent_id, - task_type="code", - ) - task_svc.start.return_value = MagicMock( - id=target_id, status="in_progress", plan={"x": 1}, assigned_to=agent_id - ) - deps = _make_deps(task=task_svc) - c = Choreographer(deps) - - env = await c.i_will_work_on( - agent_id, - target_id, - plan=_GOOD_PLAN, - steps=_STEPS, - technical_considerations=_GOOD_TC, - risks=_GOOD_RISKS, - ) - assert env.error is None - task_svc.claim.assert_awaited_once_with(target_id, agent_id) - - -@pytest.mark.asyncio -async def test_root_task_no_sequence_check() -> None: - """Root tasks (no parent) skip the sequence check entirely.""" - agent_id = uuid4() - target_id = uuid4() - target = MagicMock( - id=target_id, - status="pending", - plan={"x": 1}, - assigned_to=None, - parent_task_id=None, - sequence=5, - task_type="code", - team="backend", - ) - task_svc = _task_svc_with(target, agent_id=agent_id) - task_svc.claim.return_value = MagicMock( - id=target_id, status="claimed", plan={"x": 1}, assigned_to=agent_id - ) - task_svc.start.return_value = MagicMock( - id=target_id, status="in_progress", plan={"x": 1}, assigned_to=agent_id - ) - deps = _make_deps(task=task_svc) - c = Choreographer(deps) - - env = await c.i_will_work_on( - agent_id, - target_id, - plan=_GOOD_PLAN, - steps=_STEPS, - technical_considerations=_GOOD_TC, - risks=_GOOD_RISKS, - ) - assert env.error is None - # Sequence check should not have queried siblings on a root task - task_svc.get_subtasks.assert_not_awaited() - - # --------------------------------------------------------------------------- # A.2 ALREADY_ACTIVE # --------------------------------------------------------------------------- diff --git a/tests/unit/gateway/test_choreographer_dev.py b/tests/unit/gateway/test_choreographer_dev.py index 125b7061..21f14db6 100644 --- a/tests/unit/gateway/test_choreographer_dev.py +++ b/tests/unit/gateway/test_choreographer_dev.py @@ -3,6 +3,7 @@ from __future__ import annotations from datetime import UTC, datetime +from typing import Any from unittest.mock import AsyncMock, MagicMock from uuid import uuid4 @@ -694,3 +695,92 @@ async def test_i_am_idle_clean_returns_idle() -> None: env = await c.i_am_idle(agent_id) assert env.status == "idle" task_svc.mark_agent_idle.assert_awaited_once_with(agent_id) + + +def _passing_i_am_done_task(agent_id: Any, task_id: Any) -> Any: + """A task that clears every i_am_done gate (so the flow reaches the push).""" + return MagicMock( + id=task_id, + status="in_progress", + assigned_to=agent_id, + plan={"x": 1}, + branch_name="feature/backend/abc", + work_session_id=uuid4(), + self_verified=False, + progress_updates=[{"message": "p"}], + acceptance_criteria=["AC1"], + acceptance_criteria_status=[ + {"criterion": "AC1", "referencing_artifact_id": "c1"} + ], + commits=[{"sha": "abc"}], + pr_number=8, + pr_url="https://x/pr/8", + team="backend", + documents=[], + dev_notes="", + qa_notes="", + ) + + +def _passing_i_am_done_deps(task: Any, **overrides: AsyncMock) -> ChoreographerDeps: + """Task + journal mocks set up so i_am_done passes through to the push.""" + task_svc = AsyncMock() + task_svc.get.return_value = task + task_svc.agent_for.return_value = MagicMock( + id=task.assigned_to, role="developer", team="backend", slug=None + ) + task_svc.submit_verification.return_value = task + task_svc.submit_qa.return_value = task + task_svc.submit_for_qa.return_value = task + journal_svc = AsyncMock() + journal_svc.has_reflect_for_task.return_value = True + journal_svc.has_decision_for_task.return_value = True + journal_svc.latest_decision_at.return_value = datetime.now(UTC) + journal_svc.has_learning_for_task.return_value = False + journal_svc.has_struggle_for_task.return_value = False + return _make_deps(task=task_svc, journal=journal_svc, **overrides) + + +@pytest.mark.asyncio +async def test_i_am_done_pushes_branch_before_qa_handoff() -> None: + """i_am_done pushes the task branch so QA reviews the latest commits. + + A fix committed during a revision cycle is local-only until pushed; without + this push QA re-reviews the stale remote and re-fails the task every cycle. + """ + agent_id = uuid4() + task_id = uuid4() + git_svc = AsyncMock() + git_svc.push_task_branch.return_value = 1 + deps = _passing_i_am_done_deps( + _passing_i_am_done_task(agent_id, task_id), git=git_svc + ) + c = Choreographer(deps) + + env = await c.i_am_done(agent_id, task_id, "done") + + assert env.error is None + git_svc.push_task_branch.assert_awaited_once_with(agent_id, task_id) + + +@pytest.mark.asyncio +async def test_i_am_done_blocks_when_branch_push_fails() -> None: + """A failed push aborts i_am_done — a task must not reach awaiting_qa with + commits that live only in the developer's local workspace.""" + agent_id = uuid4() + task_id = uuid4() + git_svc = AsyncMock() + git_svc.push_task_branch.side_effect = RuntimeError("fetch timed out") + deps = _passing_i_am_done_deps( + _passing_i_am_done_task(agent_id, task_id), git=git_svc + ) + c = Choreographer(deps) + + env = await c.i_am_done(agent_id, task_id, "done") + body = env.as_dict() + + assert body["error"] == "invalid_state" + assert "push" in body["message"].lower() + # The QA transition must not have run. + deps.task.submit_qa.assert_not_awaited() + deps.task.submit_for_qa.assert_not_awaited() diff --git a/tests/unit/gateway/test_choreographer_pm.py b/tests/unit/gateway/test_choreographer_pm.py index fed6a100..27e22109 100644 --- a/tests/unit/gateway/test_choreographer_pm.py +++ b/tests/unit/gateway/test_choreographer_pm.py @@ -247,6 +247,35 @@ async def test_unblock_restore_false_returns_legacy_message() -> None: assert "re-engage" in body["next"].lower() +@pytest.mark.asyncio +async def test_unblock_refused_while_a_dependency_is_unfinished() -> None: + """A dependency block can't be force-cleared by a PM. + + It auto-clears via _unblock_dependents once the upstream completes; manual + unblock would let the dependent proceed without the upstream's work. + """ + pm_id = uuid4() + task_id = uuid4() + dep_id = uuid4() + t = MagicMock(id=task_id, status="blocked", dependency_ids=[dep_id]) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.unmet_dependency_ids.return_value = [dep_id] + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = True + journal_svc.latest_decision_at.return_value = datetime.now(UTC) + deps = _make_deps(task=task_svc, journal=journal_svc) + c = Choreographer(deps) + + env = await c.unblock(pm_id, task_id) + body = env.as_dict() + + assert body["error"] == "invalid_state" + assert "depends on" in body["message"] + # The task must not have been advanced out of blocked. + task_svc.unblock_with_restore.assert_not_awaited() + + @pytest.mark.asyncio async def test_cell_pm_complete_merges_then_completes() -> None: pm_id = uuid4() diff --git a/tests/unit/gateway/test_claim_guards_direct.py b/tests/unit/gateway/test_claim_guards_direct.py deleted file mode 100644 index d59fbde2..00000000 --- a/tests/unit/gateway/test_claim_guards_direct.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Direct unit tests for claim_guards helpers (branches only).""" - -from __future__ import annotations - -from types import SimpleNamespace -from uuid import uuid4 - -from roboco.services.gateway.claim_guards import sibling_sequence_guard - - -def test_sibling_sequence_guard_root_task_passes() -> None: - """parent_task_id None → no guard.""" - task = SimpleNamespace(id=uuid4(), parent_task_id=None, sequence=5) - assert sibling_sequence_guard(task, []) is None - - -def test_sibling_sequence_guard_sequence_zero_passes() -> None: - """sequence==0 always allowed.""" - task = SimpleNamespace(id=uuid4(), parent_task_id=uuid4(), sequence=0) - assert sibling_sequence_guard(task, []) is None - - -def test_sibling_sequence_guard_blocks_when_earlier_sibling_open() -> None: - parent = uuid4() - target = SimpleNamespace(id=uuid4(), parent_task_id=parent, sequence=2) - earlier = SimpleNamespace( - id=uuid4(), parent_task_id=parent, sequence=1, status="in_progress" - ) - env = sibling_sequence_guard(target, [earlier]) - assert env is not None - - -def test_sibling_sequence_guard_passes_when_earlier_sibling_terminal() -> None: - parent = uuid4() - target = SimpleNamespace(id=uuid4(), parent_task_id=parent, sequence=2) - earlier = SimpleNamespace( - id=uuid4(), parent_task_id=parent, sequence=1, status="completed" - ) - assert sibling_sequence_guard(target, [earlier]) is None diff --git a/tests/unit/runtime/test_stale_claim_reaper.py b/tests/unit/runtime/test_stale_claim_reaper.py index dedc4f83..0248e64c 100644 --- a/tests/unit/runtime/test_stale_claim_reaper.py +++ b/tests/unit/runtime/test_stale_claim_reaper.py @@ -18,7 +18,9 @@ from unittest.mock import AsyncMock from uuid import uuid4 import pytest -from roboco.runtime.orchestrator import AgentOrchestrator +from roboco.models.runtime import AgentInstance +from roboco.runtime.orchestrator import AgentOrchestrator, AgentState +from roboco.seeds.initial_data import AGENT_UUIDS @pytest.mark.asyncio @@ -90,3 +92,51 @@ async def test_reap_stale_claims_swallows_unclaim_errors() -> None: # Both stale tasks attempted; second succeeded despite first raising. expected_attempts = 2 assert svc.unclaim_for_reaper.await_count == expected_attempts + + +@pytest.mark.asyncio +async def test_reap_spares_claims_whose_assignee_container_is_alive() -> None: + """A stale-heartbeat task is NOT reaped while its assignee container lives. + + A developer deep in a long edit/test cycle outruns the heartbeat TTL; the + running container is the ground truth, so the claim survives rather than + being churned out from under live work. A peer task whose assignee has no + live instance is still reaped. + """ + now = datetime.now(UTC) + live_id = uuid4() + dead_id = uuid4() + live_task = type( + "T", + (), + { + "id": live_id, + "last_heartbeat_at": now - timedelta(seconds=600), + "assigned_to": AGENT_UUIDS["be-dev-1"], + "claimed_by": None, + }, + )() + dead_task = type( + "T", + (), + { + "id": dead_id, + "last_heartbeat_at": now - timedelta(seconds=600), + "assigned_to": AGENT_UUIDS["be-dev-2"], + "claimed_by": None, + }, + )() + + orch = AgentOrchestrator.__new__(AgentOrchestrator) + orch._claim_heartbeat_ttl = 300 + orch._instances = { + "be-dev-1": AgentInstance(agent_id="be-dev-1", state=AgentState.ACTIVE) + } + svc = AsyncMock() + svc.list_in_progress_or_claimed.return_value = [live_task, dead_task] + svc.unclaim_for_reaper = AsyncMock() + + await orch._reap_with_service(svc) + + # The live-assignee task is spared; only the dead one is reaped. + svc.unclaim_for_reaper.assert_awaited_once_with(dead_id) diff --git a/tests/unit/services/test_escalation_board_guard.py b/tests/unit/services/test_escalation_board_guard.py index 7a78aaa1..6f525821 100644 --- a/tests/unit/services/test_escalation_board_guard.py +++ b/tests/unit/services/test_escalation_board_guard.py @@ -12,11 +12,12 @@ stranded on a board role. from __future__ import annotations -from unittest.mock import AsyncMock, MagicMock +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch from uuid import uuid4 import pytest -from roboco.models.base import AgentRole, TaskStatus, TaskType +from roboco.models.base import AgentRole, TaskStatus, TaskType, Team from roboco.services.task import TaskService, _is_descendant_executable_task @@ -327,3 +328,44 @@ async def test_is_board_advisory_agent_classifies_roles() -> None: session.execute = AsyncMock(return_value=result) svc = TaskService(session) assert await svc._is_board_advisory_agent(uuid4()) is expected + + +@pytest.mark.asyncio +async def test_apply_escalation_emits_blocked_audit_event() -> None: + """A non-divert escalation sets BLOCKED and MUST record a task.blocked audit + row. The escalate path sets status directly (bypassing the validated + transition), and used to skip the audit log entirely.""" + svc = _service() + task = MagicMock( + id=uuid4(), + parent_task_id=uuid4(), + task_type=TaskType.PLANNING, # not cell-executed → never diverted + assigned_to=uuid4(), + claimed_by=uuid4(), + blocker_raised_by=None, + dev_notes="", + team=Team.BACKEND, + status=TaskStatus.IN_PROGRESS, + ) + _bind(svc, "_is_board_advisory_agent", AsyncMock(return_value=False)) + audit_mock = MagicMock(log_task_event=AsyncMock()) + + with patch("roboco.services.audit.get_audit_service", return_value=audit_mock): + await svc.apply_escalation( + task=task, + target_agent_id=uuid4(), + escalator_slug="be-pm", + target_slug="main-pm", + reason="needs a decision", + ) + # Drain the fire-and-forget audit task so the assertion sees the call. + pending = list(svc._background_tasks) + if pending: + await asyncio.gather(*pending, return_exceptions=True) + + assert task.status == TaskStatus.BLOCKED + audit_mock.log_task_event.assert_awaited_once() + kwargs = audit_mock.log_task_event.await_args.kwargs + assert kwargs["event_type"] == "task.blocked" + assert kwargs["details"]["from_status"] == "in_progress" + assert kwargs["details"]["to_status"] == "blocked" diff --git a/tests/unit/services/test_git.py b/tests/unit/services/test_git.py index ee2d51d1..4a6aed7a 100644 --- a/tests/unit/services/test_git.py +++ b/tests/unit/services/test_git.py @@ -22,6 +22,7 @@ if TYPE_CHECKING: from contextlib import AbstractContextManager _EXPECTED_PR_NUMBER = 7 +_PUSHED_COMMIT_COUNT = 2 def _make_session(execute_returns: object | None = None) -> MagicMock: @@ -126,6 +127,46 @@ async def test_project_for_task_uses_project_id_when_present() -> None: assert out is fake_project +# --------------------------------------------------------------------------- +# push_task_branch: idempotent push at the QA-submission boundary +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_push_task_branch_resolves_workspace_and_pushes() -> None: + """Resolves the task's project + workspace, then pushes; returns the count.""" + task = MagicMock(branch_name="feature/backend/abc") + project = MagicMock(slug="roboco") + svc = _service() + _bind(svc, "_assert_task_owned_with_branch", AsyncMock(return_value=task)) + _bind(svc, "_project_for_task", AsyncMock(return_value=project)) + _bind(svc, "get_workspace", AsyncMock(return_value=Path("/tmp/ws"))) + _bind(svc, "_assert_on_task_branch", AsyncMock()) + push_mock = AsyncMock(return_value=("feature/backend/abc", _PUSHED_COMMIT_COUNT)) + _bind(svc, "push", push_mock) + + pushed = await svc.push_task_branch(uuid4(), uuid4()) + + assert pushed == _PUSHED_COMMIT_COUNT + push_mock.assert_awaited_once_with(Path("/tmp/ws")) + + +@pytest.mark.asyncio +async def test_push_task_branch_noop_for_project_less_task() -> None: + """A git-exempt task (no resolvable project) is a no-op, not an error.""" + task = MagicMock(branch_name="feature/main_pm/abc") + svc = _service() + _bind(svc, "_assert_task_owned_with_branch", AsyncMock(return_value=task)) + _bind(svc, "_project_for_task", AsyncMock(return_value=None)) + push_mock = AsyncMock() + _bind(svc, "push", push_mock) + + pushed = await svc.push_task_branch(uuid4(), uuid4()) + + assert pushed == 0 + push_mock.assert_not_awaited() + + # --------------------------------------------------------------------------- # diff: derives parent + invokes git diff # --------------------------------------------------------------------------- @@ -404,3 +445,80 @@ async def test_create_branch_idempotent_when_branch_already_exists() -> None: assert ["checkout", "-b", branch] in calls, "checkout -b attempted" assert ["checkout", branch] in calls, "fell back to existing branch on 128" + + +def _create_branch_stubs(svc: GitService) -> None: + object.__setattr__(svc, "_resolve_base_branch", AsyncMock(return_value="master")) + object.__setattr__(svc, "_project_default_branch", AsyncMock(return_value="master")) + object.__setattr__(svc, "_token_for_project", AsyncMock(return_value=None)) + object.__setattr__( + svc, "_checkout_base_with_fallback", AsyncMock(return_value="master") + ) + + +async def _run_create_branch_with_existing_branch( + svc: GitService, branch: str, unique_commits: str +) -> list[list[str]]: + """Drive create_branch where `checkout -b` fails (branch exists) and the + branch has `unique_commits` commits of its own. Returns the git argv calls. + """ + calls: list[list[str]] = [] + + async def fake_run_git( + _workspace: object, args: list[str], **_kw: object + ) -> object: + calls.append(list(args)) + if list(args[:2]) == ["checkout", "-b"]: + return MagicMock(stdout="", returncode=1) # branch already exists + if list(args[:2]) == ["rev-list", "--count"]: + return MagicMock(stdout=f"{unique_commits}\n", returncode=0) + return MagicMock(stdout="", returncode=0) + + object.__setattr__(svc, "_run_git", fake_run_git) + with ( + patch("roboco.services.git.build_branch_name", AsyncMock(return_value=branch)), + patch( + "roboco.services.git.get_task_service", + MagicMock(return_value=MagicMock(update=AsyncMock())), + ), + ): + await svc.create_branch( + Path("/tmp/ws"), + "frontend", + GitCreateBranchRequest( + project_slug="roboco-panel", + task_id=uuid4(), + branch_type="feature", + agent_id=str(uuid4()), + parent_branch=None, + ), + ) + return calls + + +@pytest.mark.asyncio +async def test_create_branch_refreshes_no_work_existing_branch_to_base() -> None: + """An existing branch with no commits of its own is re-pointed at the fresh + base — a dependency-blocked task re-claimed after its upstream merged must + not keep building on the stale snapshot.""" + svc = _service() + _create_branch_stubs(svc) + calls = await _run_create_branch_with_existing_branch( + svc, "feature/frontend/abc12345--def67890", unique_commits="0" + ) + assert ["reset", "--hard", "master"] in calls, ( + "a no-work existing branch must be reset onto the fresh base" + ) + + +@pytest.mark.asyncio +async def test_create_branch_keeps_existing_branch_that_has_work() -> None: + """An existing branch carrying its own commits is NOT reset (work preserved).""" + svc = _service() + _create_branch_stubs(svc) + calls = await _run_create_branch_with_existing_branch( + svc, "feature/frontend/abc12345--def67890", unique_commits="3" + ) + assert not any(c[:2] == ["reset", "--hard"] for c in calls), ( + "a branch with real work must never be reset" + ) diff --git a/tests/unit/services/test_task.py b/tests/unit/services/test_task.py index 4b03c8ab..86f3dba1 100644 --- a/tests/unit/services/test_task.py +++ b/tests/unit/services/test_task.py @@ -679,3 +679,35 @@ async def test_ensure_branch_raises_when_neither_project_nor_product() -> None: task = MagicMock(branch_name=None, project_id=None, product_id=None) with pytest.raises(ValueError, match="project_id"): await svc._ensure_branch_for_task(task, uuid4()) + + +# --------------------------------------------------------------------------- +# _resolve_doc_abspath — normalize documenter-supplied paths under /app/docs +# --------------------------------------------------------------------------- + + +def test_resolve_doc_abspath_strips_redundant_docs_prefix() -> None: + """A `docs/`-rooted relative path must not double the base segment. + + DOCS_BASE_PATH is /app/docs; joining it with `docs/design/x.md` produced + /app/docs/docs/design/x.md, so the file was never found and never indexed. + """ + assert ( + TaskService._resolve_doc_abspath("docs/design/spec.md") + == "/app/docs/design/spec.md" + ) + + +def test_resolve_doc_abspath_keeps_plain_relative_path() -> None: + """A relative path with no `docs/` prefix joins under the base unchanged.""" + assert ( + TaskService._resolve_doc_abspath("design/spec.md") == "/app/docs/design/spec.md" + ) + + +def test_resolve_doc_abspath_passes_absolute_path_through() -> None: + """An already-absolute path is trusted as-is (no re-rooting).""" + assert ( + TaskService._resolve_doc_abspath("/app/docs/design/spec.md") + == "/app/docs/design/spec.md" + ) diff --git a/tests/unit/services/test_workspace_refresh.py b/tests/unit/services/test_workspace_refresh.py index 6da0c493..993a5ead 100644 --- a/tests/unit/services/test_workspace_refresh.py +++ b/tests/unit/services/test_workspace_refresh.py @@ -106,16 +106,65 @@ async def test_ensure_workspace_fetches_origin_on_healthy_short_circuit( f"Expected `git fetch origin` on healthy short-circuit, " f"got subprocess calls: {captured}" ) - # Specifically: `git fetch origin` with NO `-c` flag and no extra - # positional refspec. The `-c` check protects the docstring's + # Specifically: a SCOPED `git fetch --no-tags --prune origin ` with + # NO `-c` flag. The fetch is scoped to the workspace's branches (current + + # default) rather than all refs so it can't time out on a monorepo with many + # accumulated feature/* branches. The `-c` check protects the docstring's # no-token-injection invariant — a future refactor that added - # `git -c http.extraheader=...` would still satisfy a loose - # `a[-2:] == ["fetch", "origin"]` assertion, silently regressing - # the no-PAT-injection guarantee. + # `git -c http.extraheader=...` must not slip in unnoticed. assert any( - a[0] == "git" and "-c" not in a and a[-2:] == ["fetch", "origin"] + a[0] == "git" + and "-c" not in a + and "fetch" in a + and "--no-tags" in a + and "--prune" in a + and "origin" in a + and a.index("origin") < len(a) - 1 # ≥1 ref after origin → scoped for a in fetch_calls - ), f"Expected exact `git fetch origin` (no `-c`), got: {fetch_calls}" + ), f"Expected scoped `git fetch --no-tags --prune origin `, got: {fetch_calls}" + + +@pytest.mark.asyncio +async def test_refresh_fetch_is_scoped_to_current_and_default_branch( + healthy_workspace: Path, +) -> None: + """The refresh fetch targets only the current branch + default, not all refs. + + An all-refs fetch times out on a monorepo with many accumulated feature/* + branches, leaving the workspace silently stale. + """ + svc = _service() + agent = _fake_agent() + _bind(svc, "_lookup_agent_or_raise", AsyncMock(return_value=agent)) + _bind(svc, "get_workspace_path", MagicMock(return_value=healthy_workspace)) + + captured: list[list[str]] = [] + + def _fake_run( + args: list[str], **_kwargs: object + ) -> subprocess.CompletedProcess[str]: + captured.append(args) + out = "" + if "rev-parse" in args: + out = "feature/frontend/abc12345" + elif "symbolic-ref" in args: + out = "origin/master" + return subprocess.CompletedProcess( + args=args, returncode=0, stdout=out, stderr="" + ) + + with ( + patch("roboco.services.workspace.subprocess.run", side_effect=_fake_run), + patch("roboco.services.workspace._ensure_agent_owned"), + ): + await svc.ensure_workspace(project_slug="roboco", agent_id=agent.id) + + fetch = next(a for a in captured if a[0] == "git" and "fetch" in a) + after_origin = fetch[fetch.index("origin") + 1 :] + assert "feature/frontend/abc12345" in after_origin, ( + f"current branch must be fetched, got: {fetch}" + ) + assert "master" in after_origin, f"default branch must be fetched, got: {fetch}" @pytest.mark.asyncio diff --git a/tests/unit/test_agents_config.py b/tests/unit/test_agents_config.py index 2758f529..9718c7a6 100644 --- a/tests/unit/test_agents_config.py +++ b/tests/unit/test_agents_config.py @@ -24,8 +24,10 @@ from roboco.agents_config import ( is_management, is_pm, issue_agent_token, + issue_panel_token, verify_agent_token, ) +from roboco.seeds.initial_data import CEO_AGENT_ID if TYPE_CHECKING: import pytest @@ -295,6 +297,39 @@ def test_verify_agent_token_rejects_mismatched_signature( assert verify_agent_token(tok, "be-dev-1", "qa", "backend") is False +# --------------------------------------------------------------------------- +# issue_panel_token — the panel's CEO credential for secure mode +# --------------------------------------------------------------------------- + + +def test_issue_panel_token_verifies_under_panel_headers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The panel token must verify under the EXACT headers the panel sends: + X-Agent-Id = CEO uuid, X-Agent-Role = ceo, and NO team (empty).""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", "panel-secret") + tok = issue_panel_token() + assert verify_agent_token(tok, CEO_AGENT_ID, "ceo", "") is True + + +def test_issue_panel_token_unsigned_without_secret( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("ROBOCO_AGENT_AUTH_SECRET", raising=False) + assert issue_panel_token() == "UNSIGNED" + + +def test_panel_token_does_not_grant_other_roles_or_identities( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The panel token is bound to the CEO identity — it cannot be replayed to + claim a different role or agent id.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", "panel-secret") + tok = issue_panel_token() + assert verify_agent_token(tok, CEO_AGENT_ID, "developer", "") is False + assert verify_agent_token(tok, "be-dev-1", "ceo", "") is False + + # --------------------------------------------------------------------------- # get_pm_for_agent main_pm escalation (line 360) # ---------------------------------------------------------------------------