External/roboco
1
0
Fork 0
mirror of https://github.com/rennf93/roboco.git synced 2026-08-03 07:23:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

Feature: prompter gold upgrade (#84)

Browse Source 
* feat(prompter): make the assistant a RoboCo insider and fully wire launch

The Prompter's intelligence lived in two thin static prompts, so it asked
generic checklist questions and produced a flat task. The launch path was
also only half-wired: the panel called the generic task-create endpoint with
no project, bypassing the Prompter's own confirm flow.

Interview brain
- Rewrite the chat system prompt with RoboCo's org model, the task-spec
  standard, a dimensions playbook, and a reflect-back, 1-2-questions-per-turn,
  auto-stop discipline.
- Inject the live projects/products list each turn so the assistant grounds
  questions in real surfaces and resolves the target itself.
- Replace the brittle phrase-match readiness with a parsed roboco-meta control
  block (parse_readiness); the block is stripped from the visible reply and the
  turn now returns draft_ready + scale.

Structured GOLD draft
- Add first-class draft fields (objective, what_this_builds, the_work, notes)
  carried in the existing draft_data JSONB — no migration.
- Compose the GOLD markdown description deterministically from those fields
  (compose_description); the model never hand-formats the body.

Adaptive routing + wired launch
- Confirm now runs through the Prompter confirm endpoint with the human's
  project/product choice and edited structured draft.
- Single-cell targets a project and the cell team; a multi-cell feature targets
  a product and becomes a Main-PM coordination root that fans out.

Frontend
- Turn-envelope draft_ready (drop the duplicated phrase-match), structured
  draft card, confirm dialog with a project/product picker and a per-cell
  The Work editor, and the corrected priority labels (0 highest .. 3 lowest).

* fix(prompter): commit session writes so they survive across requests

Session create returned 201 but the row was never durably committed, so the
immediately-following /messages call could not find it and 404'd. The prompter
routes were the only write surface that never called db.commit() — every other
write route (tasks, a2a, groups, docs, product) commits explicitly rather than
rely on the request-teardown auto-commit, which is sensitive to middleware and
teardown ordering under the production server.

- Commit explicitly in all four prompter write routes (create session, send
  message, get/generate draft, confirm).
- Fix _get_session's NotFoundError: it passed a full sentence as resource_type,
  producing the doubled "... not found not found" message; now uses the
  (resource_type, resource_id) signature.
- Panel: when a message hits a session the server no longer has, start a fresh
  session and retry once instead of dead-ending on a stale id.

Add a regression test that gives each request its own non-committing session —
the real cross-request boundary the shared-session integration tests never
crossed. It reproduces the production 404 without the route commit and passes
with it.

* refactor(prompter): drop the "GOLD" jargon for plain wording

"GOLD" was informal shorthand for "a good/well-formed spec" that should never
have been baked into the LLM prompts, comments, and docstrings as if it were a
defined term. Replace it everywhere with plain language ("a well-formed task",
"a complete task spec", "the markdown description", "structured spec fields").
No behaviour change.

* feat(intake): add the intake interviewer agent role (static definition)

Phase 1 of the intake-agent feature: a new first-class `prompter` role — the
intake interviewer the CEO chats with to draft a task. This commit defines the
role across every foundation layer (no runtime yet); spawning + the live
session come next.

- identity: Role.PROMPTER, RoleLevel.INTAKE (lowest authority), an AGENTS row
  (intake-1) on the board team, ROLE_LEVEL entry. Deliberately NOT in
  BOARD_ROLES — it interviews, it does not review.
- lifecycle: gets i_am_idle like every agent (its only verb); no
  delivery-lifecycle intents.
- journaling: ReadTier.OWN — isolated, reads only its own journal.
- role_config: human-only manifest — note + evidence only, no say/dm/notify/
  channels; allows_subagent=True (research), allows_write=False.
- agents_config derives it automatically and correctly excludes it from
  TASK_CREATOR_ROLES (it drafts, it never creates tasks).
- seed presentation ("Intake"); regenerated lifecycle artifacts.
- role system prompt: read the code first, single-CEO awareness, propose
  rather than interrogate — written against the failures we saw.
- docs: roster count 19 -> 20, org charts, verb-surface table, usage roster.

All foundation drift checks pass; role/manifest/permission tests green.

* feat(intake): migrate agentrole enum to add 'prompter'

ALTER TYPE agentrole ADD VALUE IF NOT EXISTS 'prompter' so the intake agent
row seeds/spawns against a migrated production DB. Forward-only (postgres
can't drop enum values), guarded for offline mode — matches migration 012.

* style(intake): ruff format the role additions

* fix(intake): unguard the agentrole migration so it renders offline

The enum-migration-parity test renders 'alembic upgrade head --sql' (offline)
and greps for ALTER TYPE ... ADD VALUE. The is_offline_mode() guard skipped
emitting it, so the parity check couldn't see 'prompter'. Drop the guard —
PG16 permits ADD VALUE in a transaction, same as migration 020's backfill.

* feat(intake): the live-session driver (Claude Agent SDK loop)

Phase 2 begins. The intake agent isn't a one-shot `claude -p`; it's a live
Claude Code session the human chats with. This driver is the container's loop:
open one long-lived claude-agent-sdk ClaudeSDKClient, then per human message
run a turn (query + receive_response) and stream its events out, keeping
conversation context in-process — verified against the real SDK (v0.2.94).

- StreamChunk + normalize(): map SDK messages (StreamEvent text deltas,
  AssistantMessage text/thinking/tool_use blocks, ResultMessage→session_id) to
  panel-facing chunks. Duck-typed, so it works on real SDK objects and on test
  fakes alike — SDK-free, fully unit-tested.
- IntakeDriver.run(): the loop, with injected session/source/sink seams; a turn
  failure surfaces as an error chunk without killing the session.
- SdkIntakeSession + build_intake_options: the only SDK-coupled code (lazy
  import; needs the live claude binary, so excluded from coverage).
- Add claude-agent-sdk dependency + mypy ignore-missing-stubs.

Relay, panel SSE, and the persistent on-demand spawn are the next steps.

* Updated uv.lock

* feat(intake): the panel<->agent live bridge (registry, routes, entrypoint, image)

Wires the live intake chat end to end (Phase 2 integration layer):

- prompter_live.py: the orchestrator-side per-session registry — open/close,
  push (agent->panel), stream (SSE drain), deliver (panel->container). In-process
  (the orchestrator is single-process). 7 unit tests.
- routes/prompter_live.py: GET /live/{id}/stream (SSE), POST /live/{id}/messages
  (deliver), POST /live/{id}/events (relay in); registered under /api/prompter.
  5 integration tests.
- agent_sdk/intake_main.py: the container entrypoint — a POST /turn receiver
  (the driver's MessageSource) + a relay-poster EventSink + the ClaudeSDKClient
  session, run concurrently. 4 unit tests on the wiring helpers.
- docker/agent-prompter.Dockerfile: FROM base, ENTRYPOINT = the driver (not the
  one-shot `claude` the other agents use).

Remaining for Phase 2: the orchestrator persistent-spawn path (scope->workspace
clone, CMD = driver, registry.open on spawn, reap-on-confirm) — the deploy-side
piece, best finalized against a buildable image.

* feat(intake): orchestrator persistent spawn + start/stop for the live chat

Add the task-free spawn path for the intake (prompter) agent: one fixed
intake-1 container running the Agent-SDK driver (image ENTRYPOINT, not
claude -p), one live session at a time.

- spawn_intake_session clones the scope's repo(s) via WorkspaceService
  (project -> one; product -> each distinct project, primary first),
  composes the intake-1 prompt, resolves the model, and builds docker run
  via _build_intake_run_cmd: no settings/hook mount (driver owns 9000),
  no MCP config, no -w; registers the live relay and best-effort delivers
  the opening message once the receiver is up.
- reap_intake_session closes the relay and stops the container.
- Routes: POST /live/start (project XOR product) and POST /live/{id}/stop.
- ROLE_MODEL_MAP[prompter]=opus; intake-1 -> roboco-agent-prompter image map.
- Replace the budget-sweep try/except/continue with _fetch_budget_status,
  which logs the swallow at debug instead of silently dropping it.

25 new tests; docker + the clone are mocked. End-to-end container spawn is
pending a built image and the stack.

* feat(intake): wire /prompter to the live agent — scope form + SSE chat

Replace the Ollama chat loop on /prompter with the spawned-agent flow.

- IntakeForm: pick scope (project XOR product) + opening message + Start
  before the chat; the agent clones that scope and reads the real code.
- use-prompter rewritten as the live brain (lib/api/prompter-live.ts): Start
  spawns via POST /live/start, then an EventSource on /live/{id}/stream
  streams the agent working — token deltas fill the assistant bubble,
  tool_use/thinking drive a live activity line, a draft event renders the
  existing DraftProposalCard. Messages go via POST /live/{id}/messages.
- Chat UX unchanged (Keep Chatting / Review & Confirm / ConfirmDialog reused);
  reap-on-confirm and reap-on-leave call POST /live/{id}/stop.
- Drop the dead Ollama prompterApi client; trim prompter.ts to shared types.

Frontend gate green (tsc --noEmit, lint, build). The draft event + the
/live/{id}/confirm endpoint are the Phase 4 backend seam.

* feat(intake): confirm draft -> backlog task + agent draft emission

Complete the live intake vertical: the agent proposes a structured draft and
Review & Confirm turns it into a task.

- Draft emission: the prompter prompt instructs the agent to emit a fenced
  roboco-draft JSON block when the spec is ready; the driver parses it into a
  'draft' event over the existing relay -> the panel's DraftProposalCard. The
  panel strips the raw block from the chat bubble.
- Fix a double-text bug: with include_partial_messages the reply arrives as
  both StreamEvent deltas and the final AssistantMessage; the driver now takes
  text from deltas only and the AssistantMessage for thinking/tool_use/draft.
- POST /live/{id}/confirm -> confirm_live_draft, reusing a draft->task core
  extracted from confirm_draft; reaps the session on success.
- Both prompter confirm paths create at BACKLOG, not pending: backlog is the
  holding area a draft waits in until it's reviewed and promoted to pending
  (TaskService.activate). The legacy Ollama confirm was creating at pending,
  skipping that gate — fixed.
- Remove the dead 'context' bootstrap param from the Ollama session-create
  chain (schema + route + method + tests), superseded by the live scope form.
- No suppressions: replace every type:ignore/noqa across the intake surface
  with a real fix (ORM .id -> UUID(str(x)); fakes -> monkeypatch.setattr;
  lazy imports -> pyproject per-file ignore; union-attr -> recipients[0]).

Full make quality green; frontend tsc + lint green.

* build(intake): add the agent-prompter image builder to compose

The orchestrator references roboco-agent-prompter (AGENT_IMAGES + the
_ensure_agent_image dockerfile map) and docker/agent-prompter.Dockerfile
exists, but docker-compose.yml built every other agent image up front and
left this one out — so the image wasn't pre-built for a stack bring-up.

Mirror the other specialized agent-*-image builders: build from
docker/agent-prompter.Dockerfile, tag roboco-agent-prompter, depend on
agent-base-image.

* Created docker-compose.yaml for the NAS

* fix(intake): non-blocking /live/start so spawn never times out

The start POST awaited the whole spawn — workspace clone + first-time image
build + docker run — which blew past the panel's 60s HTTP timeout ('Request
timed out. The server may be busy.') and triggered a duplicate send. Found on
the 2026-06-09 NAS smoke.

- start_intake_session opens the live relay synchronously, then spawns the
  container in the background (_spawn_intake_container_guarded). The route
  returns the session id immediately; the panel opens the SSE stream right away.
- A background spawn failure is pushed onto the relay as an 'error' event and
  closes the session, so the panel shows it instead of hanging.
- spawn_intake_session stays as the synchronous variant for direct callers/tests.
- Panel shows a 'Preparing the agent…' indicator until the first event arrives.

18 intake-spawn tests green; tsc + lint green. E2E re-validates on next smoke.

* fix(intake): propose_draft MCP tool + lock the agent down

Smoke 2026-06-09 exposed two compounding problems: the agent never reliably
emitted the draft (it narrated the spec instead of typing the magic fence), and
it had inherited the CEO's entire Claude Code env — Write/Edit/Bash + Gmail/
Notion/Calendar/Drive MCP — because bypassPermissions ignored the allowlist and
the mounted ~/.claude leaked the host MCP config.

- propose_draft: build_intake_options now registers an in-process SDK MCP tool
  (create_sdk_mcp_server + @tool). The agent calls it to submit the draft; the
  driver turns that ToolUseBlock into a 'draft' event (_is_propose_draft /
  _draft_from_tool_input, tolerant of nested/flat/JSON-string input). The fenced
  roboco-draft block stays as a fallback.
- Lockdown: strict_mcp_config=True + setting_sources=[] (ignore host MCP +
  settings); permission_mode 'dontAsk' + a can_use_tool gate enforcing a hard
  allowlist (Read/Grep/Glob/Task + propose_draft) replaces bypassPermissions.
- Prompt: call propose_draft (not a fence); the draft's downstream chain is
  backlog -> Board (PO + HoM) -> CEO approve -> Main PM, and the agent's job ends
  at the draft (it never routes or hands off).

SDK API verified against the installed claude-agent-sdk. Driver detection unit-
tested; the SDK-construction is validated on the next NAS smoke (incl. that
setting_sources=[] doesn't break the mounted-~/.claude auth).

* fix(intake): panel UX cluster from the smoke (#3/#4/#6/#12)

- #3 message boundaries: a tool call now ends the current text bubble, so the
  agent's words before and after a tool render as separate messages instead of
  one merged wall (the 'two waves merged into one bubble' the CEO saw).
- #4 activity indicator: promoted from tiny grey text to a prominent primary-
  tinted pill so 'watch it work' is actually visible.
- #12 End chat: a header button (any chat state) reaps the agent and resets to
  the form, reusing startAnother (which already stops the session). Backend
  POST /live/{id}/stop already existed.
- #6 log noise: the opening-message delivery retry logs at debug, not error —
  those failures are expected until the container receiver is up.
- Also fix a latent test gap from the #1 commit: the live-route test's fake
  orchestrator now exposes start_intake_session (the route's non-blocking entry).

Frontend tsc + lint green; live-route + prompter_live tests green.

* fix(intake): render markdown in the chat bubbles (#8)

The agent emits rich markdown (### headers, **bold**, tables, lists) but the
bubble rendered raw text, so it was illegible (CEO-flagged on the smoke). Render
assistant content with react-markdown + remark-gfm (GFM tables) in a prose
container. Adds react-markdown + remark-gfm to the panel.

* feat(intake): #14 — two start routes (Board review vs straight to Main PM)

Per the CEO spec, the draft confirm now starts the task at PENDING with an
explicit assignment instead of parking it at backlog:

- route="board" (Board review & Start): assigned to the Product Owner, so the
  orchestrator dispatches the full Board review (PO + Head of Marketing) before
  the Main PM picks it up.
- route="main_pm" (Approve & Start): assigned straight to the Main PM, who
  delegates to the cells (Board review skipped).

create_task_from_draft gains status + assigned_to params (default BACKLOG, so the
legacy confirm_draft is unchanged); confirm_live_draft + the /live/{id}/confirm
request carry the route. Service tests cover both routes.

* feat(intake): #14 draft-card buttons — Board review vs Approve & Start

Three buttons on the draft card now (CEO spec): Keep chatting / Board review &
Start / Approve & Start. The two action buttons confirm directly with their
route — launchTask(route) sends route to POST /live/{id}/confirm, which starts
the task at pending assigned to the Board (PO+HoM) or straight to the Main PM.

Supersedes the ConfirmDialog review step (scope is chosen up front in the form),
so it's removed from the page flow. The ConfirmDialog component + its sub-editors
are now unused — flagged for a follow-up cleanup, left in place to avoid churn.

tsc + lint green.

* fix(intake): keep the live SSE stream bound to its relay session

The orchestrator opened the relay session twice per live chat — once on the
request path (before the start call returns) and again inside the background
container spawn. The SSE stream binds to the session's queue the moment the
panel connects, so the second open swapped in a fresh queue and stranded the
stream: the agent replied normally, but its events went to the new queue while
the panel kept reading the old one, so the chat looked frozen on "Preparing…".

The second open was always redundant (the relay is opened by the caller before
the spawn). Remove it, and make open() idempotent so a live session is never
replaced out from under a stream that is already connected to it.

* fix(intake): draft-card launch buttons silently did nothing

The launch path required a `description` field, but the prompter draft schema
intentionally has none — it sends `objective` + the structured spec and the
backend composes the description (compose_description). `editableDraft.description`
was therefore undefined, so `description.trim()` inside launch validation threw a
TypeError that propagated out of the button's onClick. Clicking "Board review &
Start" / "Approve & Start" did nothing, with no feedback — the wall blocking the
whole confirm → task → reap flow.

- Map a proposed draft's description from `objective` as a fallback.
- Make launch validation null-safe.
- Replace the silent early-return with a toast that names what's missing, so a
  blocked launch is never a dead, feedback-less button again.

* fix(intake): steer the agent to ask inline, not via AskUserQuestion

The intake's job is to ask clarifying questions, so it reached for the
AskUserQuestion tool — which isn't wired to the live chat panel and isn't in its
allowlist. The bare deny left it to stumble ("let me clarify… — no worries, let
me just lay it out") and waste a visible turn.

- Prompt: spell out that it asks by writing in the chat (the human reads every
  message live) and that no question/prompt tool is available to it.
- Gate: give AskUserQuestion a specific deny message that nudges it to ask inline,
  so even a reflex attempt degrades gracefully.

Also refresh the now-stale "what happens after propose_draft" section: the draft
card has three choices (Keep chatting / Board review & Start / Approve & Start)
and produces a pending task — not the old two-button "backlog" description.

* feat(intake): copy buttons on agent messages and the draft card

The CEO asked for a way to save the agent's plan/spec elsewhere "just in case" —
a cheap manual backstop until refresh-durability lands.

- New CopyButton: async Clipboard API when available, plus a legacy
  textarea+execCommand fallback. The fallback is load-bearing — the panel is
  served over plain http on a LAN IP, where navigator.clipboard is absent
  (clipboard needs a secure context), so the modern API alone would never copy.
- Copy button under each assistant message (copies its text).
- Copy button on the draft card (copies the full spec as markdown: title,
  objective, what-this-builds, the-work per cell, notes, success criteria).

* feat(intake): unbuffer logs + log each turn so the container isn't a black box

Debugging the intake smoke was painful for two reasons: (a) the orchestrator
block-buffered stdout, so `docker logs` lagged minutes behind reality, and (b)
the intake container logged only "session opened" then went silent for the whole
conversation (the chat streams to the relay, not stdout).

- Set PYTHONUNBUFFERED=1 on the orchestrator and agent-base images so structured
  logs reach `docker logs` in real time instead of in large delayed chunks.
- Log each intake turn: "turn received" (with char count) and "turn streamed"
  (chunk count + whether a draft was emitted), so the container logs show the
  conversation's shape at a glance.

* chore(intake): remove the dead ConfirmDialog draft editor

The three-button draft card (Keep chatting / Board review & Start / Approve &
Start) replaced the old review-modal confirm flow, leaving ConfirmDialog and its
sub-editors (StringListEditor, TheWorkEditor) referenced by nothing but the
barrel export. Remove the three files and the export — typecheck + lint confirm
no remaining references.

* fix(intake): coerce bad draft enums on confirm instead of hard-failing

The intake agent is an LLM and will emit off-enum values — e.g. task_type="feature",
which is not a valid TaskType (code/documentation/research/planning/design/
administrative). `_coerce_draft_enums` called `TaskType(value)` directly, which
raised, and the confirm 400'd with "Draft has invalid or missing required fields:
'feature' is not a valid TaskType". That forced the agent to discover the valid
values and self-correct in-chat — unacceptable: clicking "Approve & Start" must
never blow up on a cosmetic enum guess.

Coerce each enum to a sane default on invalid/missing (task_type→code,
nature→technical, complexity→medium); team falls back to the first valid cell in
the_work, then backend. `_lead_cell_team` now skips invalid cell names too. The
confirm/launch action no longer hard-fails on an enum the model got wrong.

* fix(intake): draft card no longer renders above the user's latest message

attachDraft fell back to "the last assistant message anywhere" when the current
turn had no streamed text yet (propose_draft called first). That last message was
often the PREVIOUS turn's — sitting above the user's "Yes, propose it" — so the
draft card rendered above the user's message. Attach only to the current turn's
streaming message; otherwise append a fresh assistant message so the card always
lands at the bottom of the thread.

* test(intake): guard draft enum coercion + invalid-cell skipping

Regression tests for the confirm-time enum coercion: an off-enum task_type
("feature") / nature / complexity coerce to code/technical/medium instead of
raising, and _lead_cell_team skips invalid cell names. Locks in that a bad enum
guess from the agent can never 400 the launch again.

* fix(intake): stop the agent fumbling through Claude Code meta-tools

In smoke it reflexively probed CC built-ins before reaching propose_draft —
plan mode + ExitPlanMode (it announced a written plan and waited instead of
emitting the draft), ToolSearch, Write — each correctly denied by the lockdown
but stumbly, and it only proposed after explicit CEO nudges.

- Gate: ExitPlanMode now gets a specific deny nudge ("you don't use plan mode;
  call propose_draft"), and the generic deny names the actual toolset instead
  of a bare "not available", so any probe degrades into guidance.
- Prompt: forbid plan mode/ExitPlanMode/ToolSearch explicitly and spell out
  "you do not plan and wait — call propose_draft directly when the spec is
  ready," plus an anti-pattern bullet.

* feat(intake): make the container logs transparent mid-turn

`docker logs` on the intake container was a black box: only turn start/end, while
the agent read the codebase and spawned 20+ subagents invisibly (the conversation
streams to the relay, not stdout), and the benign 3x ~/.claude.json warning was
the only thing visible.

- Driver logs each tool call mid-turn ("Intake tool use" with the tool name) and
  the draft emission, plus a tools count in the turn-streamed summary. Text deltas
  stay unlogged (they'd spam). Now the logs show the turn's real shape.
- Pre-create ~/.claude.json ({}) at container boot so the CLI's "config not found"
  warning (printed 3x, self-healed anyway) stops drowning the real logs.

* fix(intake): render markdown in user messages + scope copy to code blocks

Two display fixes from the smoke:
- User messages collapsed newlines (plain {content} in a div) and rendered no
  markdown — a "1.\n2.\n3." answer showed as one run-on line. Render user AND
  assistant bubbles through a shared GFM markdown body that inherits the bubble's
  text color, so lists / newlines / styling render correctly on both.
- Copy was blanketed on every assistant message; scope it to KEY parts — a copy
  button on fenced code blocks (the draft card keeps its own). Removed the
  per-message button.

* fix(intake): prevent duplicate tasks from a double-click on launch

Clicking a draft launch button twice fired two confirms and created duplicate
tasks. Add a synchronous re-entry guard (a ref — no stale-closure window) at the
top of launchTask so a second click returns immediately, and disable + spin the
draft-card buttons while a launch is in flight so it's visually clear it's working.

* docs(how-to): lead task creation with the Task Assistant flow

Rewrite "1 · It starts with you" to walk the Prompter/Task Assistant path —
scope form, the agent reading the codebase, its grounded analysis, the draft
card, and the created task — then flow into the Board review. Replaces the old
manual task-definition form shots.

Image placeholder: images/prompter_draft_card.png (the 3-button card) is
referenced but not yet captured — TODO comment marks it for the next smoke run.
A second comment flags an optional re-capture of prompter_run_2 after the
markdown-rendering fix.

* fix(intake): restore assistant message text contrast

The markdown refactor dropped `dark:prose-invert` and made text inherit the
bubble's color, but the assistant bubble had no explicit text color — so its text
rendered near-invisible (dark-on-dark on bg-muted). Give the assistant bubble an
explicit text-foreground; the user bubble already carries text-primary-foreground,
and [&_*]:!text-inherit now resolves to a readable color on both.

* fix(intake): coerce draft priority too — confirm 500'd on priority="high"

The enum-coercion fix covered task_type/nature/complexity/team, but priority is a
non-enum int field handled by `int(draft_data.get("priority", 2))`, and the agent
guesses a word ("high") as readily as a number — so int("high") raised ValueError
and the confirm 500'd. Same class of bug, one field missed.

Add _coerce_priority: map words (urgent/high/medium/low → 0/1/2/3), clamp numbers
to 0-3, default to 2 (medium) on anything else. The launch can no longer crash on
any field the LLM guessed. + regression test.

* fix(intake): draft card shows distinct cells, not one badge per work item

the_work has one entry per work item, so a cell with several items rendered its
badge repeatedly ("Board-led across Backend Backend Backend Frontend Frontend
…"). De-dupe to distinct teams so the card reads "Board-led across Backend
Frontend" — and the "Cell:" vs "Board-led across" label keys off distinct count.

* docs(how-to): hero the teaser gif + resolve the Prompter/Task Assistant thread

- Move the 12s teaser gif to the top as the hero — it was buried between the
  "prefer video" link and the first screenshot.
- Name the connection: the Task Assistant IS the Prompter, so section 1 (using
  the tool) and the rest (RoboCo building it) read as one story — you use the
  tool the company built for itself, then watch the build.
- Re-anchor the section 1 → Board transition to follow the Prompter's own
  journey, instead of implying section 1's example task is the one reviewed next.

* Included images for how-to.md

* docs(how-to): align agent count to 20 (matches README + CLAUDE.md)

The how-to said "18 agents" with UX/UI at one dev and no Intake — stale against
the authoritative count. Bump 18→20 (prose + spelled-out eighteen→twenty), give
UX/UI 2 devs, and add the Intake line to the org tree (Intake leads section 1, so
it belongs in the tree). README + CLAUDE.md already say 20.

* ci(release): publish all RoboCo images to GHCR + Docker Hub

The release published only the orchestrator to GHCR. Build and push the full set
the stack needs — agent-base, the 8 agent images, orchestrator, and panel — to
BOTH ghcr.io/rennf93/* and docker.io/renzof93/*, at :<version> and :latest, so
consumers can pull instead of compose-building.

- agent-base builds first (the agent images build FROM roboco-agent-base, a local
  tag), then the rest; push only after every build succeeds.
- Image names mirror the docker-compose `image:` values 1:1.
- Free disk on the runner first (11 images is space-heavy).
- Needs a DOCKERHUB_TOKEN repo secret for the Docker Hub login.
- SECURITY.md updated to reference both registries.

* ci(release): use short SHA as the image tag on manual dispatch

A workflow_dispatch runs against a branch, and the branch name (e.g.
feature/prompter-gold-upgrade) was used verbatim as the image tag — but "/" is
illegal in a Docker tag, so the first build failed instantly with "invalid
reference format". Releases still tag from the release tag; manual dispatch now
always uses the short SHA, which is a valid tag.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>
This commit is contained in:
Renzo F
2026-06-09 17:08:34 +02:00
committed by GitHub
co-authored by Renn F
parent ae65bff883
commit 9f8834155a
63 changed files with 5607 additions and 832 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/foundation/test_identity.py
+2
View File
@@ -20,6 +20,7 @@ def test_role_enum_has_every_role_inc_system() -> None:
"product_owner",
"head_marketing",
"auditor",
"prompter",
"ceo",
"system",
}
@@ -80,6 +81,7 @@ def test_agents_catalog_has_all_seed_slugs() -> None:
"product-owner",
"head-marketing",
"auditor",
"intake-1",
}
actual = set(identity.AGENTS.keys())
assert actual == expected_slugs, f"agent catalog drift: {actual ^ expected_slugs}"
tests/foundation/test_lifecycle_spec.py
+1
View File
@@ -31,6 +31,7 @@ def test_role_enum_has_every_pre_gateway_role() -> None:
"product_owner",
"head_marketing",
"auditor",
"prompter", # post-gateway intake role (human-only, drafts tasks)
"ceo",
"system",
}
tests/integration/test_prompter_live_routes.py
+332
View File
@@ -0,0 +1,332 @@
"""Integration tests for the live intake chat routes (start/stop + relay + msg).
The SSE stream generator itself is unit-tested at the service layer
(``test_prompter_live``); here we exercise the HTTP contracts against an
injected registry whose container deliveries hit a mocked transport, and the
start/stop routes against a fake orchestrator.
"""
from __future__ import annotations
from http import HTTPStatus
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any
from unittest.mock import patch
from uuid import uuid4
import httpx
import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from roboco.api import deps
from roboco.api.deps import get_agent_context
from roboco.api.routes.prompter_live import router
from roboco.db.base import get_db
from roboco.models.base import AgentRole
from roboco.services import prompter_live
from roboco.services.base import ValidationError
from roboco.services.permissions import AgentContext
if TYPE_CHECKING:
from collections.abc import AsyncIterator
@pytest_asyncio.fixture
async def live_client() -> AsyncIterator[dict[str, Any]]:
def container_handler(_req: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"ok": True})
mock_client = httpx.AsyncClient(transport=httpx.MockTransport(container_handler))
registry = prompter_live.PrompterLiveRegistry(http_client=mock_client)
prompter_live._RegistryHolder.instance = registry # inject the singleton
app = FastAPI()
app.include_router(router, prefix="/api/prompter")
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
yield {"client": client, "registry": registry}
prompter_live._RegistryHolder.instance = None
await mock_client.aclose()
@pytest.mark.asyncio
async def test_relay_event_pushes_to_live_session(live_client: dict) -> None:
client, registry = live_client["client"], live_client["registry"]
registry.open("s1", "intake-1")
resp = await client.post(
"/api/prompter/live/s1/events", json={"kind": "text", "text": "hi"}
)
assert resp.status_code == HTTPStatus.OK
assert resp.json() == {"pushed": True}
# The event is now on the session's queue.
assert registry.get("s1").queue.qsize() == 1
@pytest.mark.asyncio
async def test_relay_event_unknown_session_is_noop(live_client: dict) -> None:
resp = await live_client["client"].post(
"/api/prompter/live/nope/events", json={"kind": "text"}
)
assert resp.status_code == HTTPStatus.OK
assert resp.json() == {"pushed": False}
@pytest.mark.asyncio
async def test_send_message_delivers_to_container(live_client: dict) -> None:
client, registry = live_client["client"], live_client["registry"]
registry.open("s1", "intake-1")
resp = await client.post(
"/api/prompter/live/s1/messages", json={"text": "hello there"}
)
assert resp.status_code == HTTPStatus.OK
assert resp.json() == {"delivered": True}
@pytest.mark.asyncio
async def test_send_message_unknown_session_404(live_client: dict) -> None:
resp = await live_client["client"].post(
"/api/prompter/live/nope/messages", json={"text": "hi"}
)
assert resp.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_send_message_requires_text(live_client: dict) -> None:
live_client["registry"].open("s1", "intake-1")
resp = await live_client["client"].post(
"/api/prompter/live/s1/messages", json={"text": ""}
)
assert resp.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
# ---------------------------------------------------------------------------
# start / stop — spawn + reap against a fake orchestrator (no docker).
# ---------------------------------------------------------------------------
class _FakeOrchestrator:
"""Records spawn/reap calls; stands in for the real orchestrator singleton."""
def __init__(self) -> None:
self.spawned: list[dict[str, Any]] = []
self.reaped: list[str] = []
async def start_intake_session(
self,
session_id: str,
*,
project_slug: str | None = None,
product_id: str | None = None,
initial_message: str | None = None,
) -> None:
# The route is non-blocking now: it calls start_intake_session (returns
# None) which opens the relay + spawns in the background.
self.spawned.append(
{
"session_id": session_id,
"project_slug": project_slug,
"product_id": product_id,
"initial_message": initial_message,
}
)
async def reap_intake_session(self, session_id: str) -> None:
self.reaped.append(session_id)
@pytest_asyncio.fixture
async def start_client(
monkeypatch: pytest.MonkeyPatch,
) -> AsyncIterator[dict[str, Any]]:
orch = _FakeOrchestrator()
# monkeypatch.setattr is untyped (no ignore for the fake) and auto-reverts.
monkeypatch.setattr(deps._ServiceHolder, "orchestrator", orch)
async def _fake_db() -> AsyncIterator[object]:
yield object() # product-scope start never touches it
app = FastAPI()
app.include_router(router, prefix="/api/prompter")
app.dependency_overrides[get_db] = _fake_db
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
yield {"client": client, "orch": orch}
@pytest.mark.asyncio
async def test_start_product_scope_spawns_and_returns_session(
start_client: dict,
) -> None:
client, orch = start_client["client"], start_client["orch"]
product_id = str(uuid4())
resp = await client.post(
"/api/prompter/live/start",
json={"product_id": product_id, "initial_message": "build X"},
)
assert resp.status_code == HTTPStatus.CREATED
session_id = resp.json()["session_id"]
assert session_id
assert orch.spawned == [
{
"session_id": session_id,
"project_slug": None,
"product_id": product_id,
"initial_message": "build X",
}
]
@pytest.mark.asyncio
async def test_start_project_scope_resolves_slug(start_client: dict) -> None:
client, orch = start_client["client"], start_client["orch"]
project_id = uuid4()
fake_svc = SimpleNamespace(
get=lambda _pid: _async_return(SimpleNamespace(slug="roboco"))
)
with patch("roboco.services.project.get_project_service", lambda _db: fake_svc):
resp = await client.post(
"/api/prompter/live/start", json={"project_id": str(project_id)}
)
assert resp.status_code == HTTPStatus.CREATED
assert orch.spawned[0]["project_slug"] == "roboco"
assert orch.spawned[0]["product_id"] is None
@pytest.mark.asyncio
async def test_start_unknown_project_404(start_client: dict) -> None:
client = start_client["client"]
fake_svc = SimpleNamespace(get=lambda _pid: _async_return(None))
with patch("roboco.services.project.get_project_service", lambda _db: fake_svc):
resp = await client.post(
"/api/prompter/live/start", json={"project_id": str(uuid4())}
)
assert resp.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_start_requires_exactly_one_scope(start_client: dict) -> None:
client = start_client["client"]
both = await client.post(
"/api/prompter/live/start",
json={"project_id": str(uuid4()), "product_id": str(uuid4())},
)
assert both.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
neither = await client.post("/api/prompter/live/start", json={})
assert neither.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_stop_reaps_session(start_client: dict) -> None:
client, orch = start_client["client"], start_client["orch"]
resp = await client.post("/api/prompter/live/sess-9/stop")
assert resp.status_code == HTTPStatus.OK
assert resp.json() == {"stopped": True}
assert orch.reaped == ["sess-9"]
def _async_return(value: Any) -> Any:
"""Wrap a value in an awaitable so a lambda can stand in for an async method."""
async def _coro() -> Any:
return value
return _coro()
# ---------------------------------------------------------------------------
# confirm — draft → task + reap-on-confirm (service mocked; route wiring only).
# ---------------------------------------------------------------------------
class _FakeDb:
async def commit(self) -> None:
return None
@pytest_asyncio.fixture
async def confirm_client(
monkeypatch: pytest.MonkeyPatch,
) -> AsyncIterator[dict[str, Any]]:
orch = _FakeOrchestrator()
monkeypatch.setattr(deps._ServiceHolder, "orchestrator", orch)
async def _fake_db() -> AsyncIterator[_FakeDb]:
yield _FakeDb()
ceo = AgentContext(agent_id=uuid4(), role=AgentRole.CEO, team=None, slug="ceo")
app = FastAPI()
app.include_router(router, prefix="/api/prompter")
app.dependency_overrides[get_db] = _fake_db
app.dependency_overrides[get_agent_context] = lambda: ceo
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
yield {"client": client, "orch": orch}
@pytest.mark.asyncio
async def test_confirm_creates_task_and_reaps(confirm_client: dict) -> None:
client, orch = confirm_client["client"], confirm_client["orch"]
task_id = uuid4()
class _FakeService:
async def confirm_live_draft(self, _draft: Any, _agent: Any, **_kw: Any) -> Any:
return task_id
with patch(
"roboco.api.routes.prompter_live.get_prompter_service",
lambda _db: _FakeService(),
):
resp = await client.post(
"/api/prompter/live/s1/confirm",
json={
"project_id": str(uuid4()),
"draft": {"title": "x", "acceptance_criteria": ["a"]},
},
)
assert resp.status_code == HTTPStatus.CREATED
assert resp.json() == {"task_id": str(task_id)}
assert orch.reaped == ["s1"] # reap-on-confirm
@pytest.mark.asyncio
async def test_confirm_requires_exactly_one_target(confirm_client: dict) -> None:
client = confirm_client["client"]
both = await client.post(
"/api/prompter/live/s1/confirm",
json={
"project_id": str(uuid4()),
"product_id": str(uuid4()),
"draft": {"title": "x"},
},
)
assert both.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_confirm_validation_error_is_translated_and_not_reaped(
confirm_client: dict,
) -> None:
client, orch = confirm_client["client"], confirm_client["orch"]
class _FakeService:
async def confirm_live_draft(self, _draft: Any, _agent: Any, **_kw: Any) -> Any:
raise ValidationError(message="bad draft", field="title")
with patch(
"roboco.api.routes.prompter_live.get_prompter_service",
lambda _db: _FakeService(),
):
resp = await client.post(
"/api/prompter/live/s1/confirm",
json={"project_id": str(uuid4()), "draft": {"title": "x"}},
)
assert resp.status_code == HTTPStatus.BAD_REQUEST
assert orch.reaped == [] # a failed confirm must NOT reap the session
tests/integration/test_prompter_routes.py
+109 -23
View File
@@ -25,6 +25,7 @@ from roboco.api.routes.prompter import router as prompter_router
from roboco.db.tables import AgentTable, ProjectTable
from roboco.models.base import AgentRole, AgentStatus, Team
from roboco.models.permissions import AgentContext
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
if TYPE_CHECKING:
from collections.abc import AsyncIterator
@@ -44,8 +45,9 @@ _DOUBLE_TURN_MSGS = 4 # 2 user + 2 assistant
async def prompter_client(
db_session: AsyncSession,
) -> AsyncIterator[dict[str, Any]]:
agent_id = uuid4()
agent = AgentTable(
id=uuid4(),
id=agent_id,
name="DevAgent",
slug=f"dev-agent-{uuid4().hex[:8]}",
role=AgentRole.DEVELOPER,
@@ -68,7 +70,7 @@ async def prompter_client(
async def _override_agent() -> AgentContext:
return AgentContext(
agent_id=agent.id, # type: ignore[arg-type]
agent_id=agent_id,
role=AgentRole.DEVELOPER,
team=None,
)
@@ -118,6 +120,94 @@ async def project_fixture(db_session: AsyncSession) -> ProjectTable:
_HDR = {"X-Agent-ID": "be-dev-1", "X-Agent-Role": "developer"}
@pytest_asyncio.fixture
async def cross_request_client(
_test_database_url: str,
) -> AsyncIterator[dict[str, Any]]:
"""Client whose DB dependency yields a fresh, NON-auto-committing session
per request.
This is the boundary the shared-session ``prompter_client`` fixture can't
exercise: here a write is only visible to the next request if the route
committed it explicitly. The seed agent is committed up front so both
requests can resolve it.
"""
engine = create_async_engine(_test_database_url, future=True, pool_pre_ping=True)
maker = async_sessionmaker(bind=engine, expire_on_commit=False)
agent_id = uuid4()
async with maker() as seed:
seed.add(
AgentTable(
id=agent_id,
name="XReqAgent",
slug=f"xreq-agent-{uuid4().hex[:8]}",
role=AgentRole.DEVELOPER,
team=None,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="dev",
capabilities=[],
permissions={},
metrics={},
)
)
await seed.commit()
app = FastAPI()
app.include_router(prompter_router, prefix="/api/prompter")
async def _override_db() -> AsyncIterator[AsyncSession]:
# A fresh session per request that does NOT commit on teardown, so
# persistence depends solely on the route's explicit commit.
async with maker() as session:
yield session
async def _override_agent() -> AgentContext:
return AgentContext(agent_id=agent_id, role=AgentRole.DEVELOPER, team=None)
app.dependency_overrides[get_db] = _override_db
app.dependency_overrides[get_agent_context] = _override_agent
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
yield {"client": client, "agent_id": agent_id}
app.dependency_overrides.clear()
await engine.dispose()
@pytest.mark.asyncio
async def test_session_persists_across_requests(cross_request_client: dict) -> None:
"""A created session must survive into the next request's own DB session.
Regression for the production 404: the create returned 201 but the session
write was never committed, so the immediately-following /messages call could
not find it. Without the route's explicit commit, this is a 404.
"""
client = cross_request_client["client"]
create = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
assert create.status_code == HTTPStatus.CREATED
session_id = create.json()["id"]
reply = (
'ack\n```roboco-meta\n{"covered": [], "ready": false, "scale": "single"}\n```'
)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=reply,
):
msg = await client.post(
f"/api/prompter/sessions/{session_id}/messages",
json={"content": "hello"},
headers=_HDR,
)
assert msg.status_code == HTTPStatus.OK, msg.json()
assert len(msg.json()["messages"]) == _SINGLE_TURN_MSGS
# =============================================================================
# Session-based endpoint tests
# =============================================================================
@@ -140,20 +230,6 @@ async def test_create_session_success(prompter_client: dict) -> None:
assert "created_at" in body
@pytest.mark.asyncio
async def test_create_session_with_context(prompter_client: dict) -> None:
"""POST /sessions accepts optional bootstrap context."""
client = prompter_client["client"]
response = await client.post(
"/api/prompter/sessions",
json={"context": {"team": "backend", "project_id": str(uuid4())}},
headers=_HDR,
)
assert response.status_code == HTTPStatus.CREATED
body = response.json()
assert body["status"] == "active"
@pytest.mark.asyncio
async def test_send_message_success(prompter_client: dict) -> None:
"""POST /sessions/{id}/messages appends user+assistant messages."""
@@ -177,24 +253,28 @@ async def test_send_message_success(prompter_client: dict) -> None:
)
assert response.status_code == HTTPStatus.OK
messages = response.json()
body = response.json()
messages = body["messages"]
assert len(messages) == _SINGLE_TURN_MSGS
roles = [m["role"] for m in messages]
assert "user" in roles
assert "assistant" in roles
assert messages[-1]["content"] == "Great! Let's gather requirements."
assert body["draft_ready"] is False
@pytest.mark.asyncio
async def test_send_message_marks_draft_ready(prompter_client: dict) -> None:
"""draft_ready signal in LLM response updates session status."""
"""A ready roboco-meta control block flips draft_ready and session status."""
client = prompter_client["client"]
session_resp = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
session_id = session_resp.json()["id"]
mock_response = (
"I have enough information to draft a task now. Ready to draft when you are."
"Understood — I have what I need.\n\n"
'```roboco-meta\n{"covered": ["objective", "scope", "surface", '
'"acceptance"], "ready": true, "scale": "single"}\n```'
)
with patch(
@@ -209,8 +289,12 @@ async def test_send_message_marks_draft_ready(prompter_client: dict) -> None:
)
assert response.status_code == HTTPStatus.OK
messages = response.json()
assert len(messages) == _SINGLE_TURN_MSGS
body = response.json()
assert len(body["messages"]) == _SINGLE_TURN_MSGS
assert body["draft_ready"] is True
assert body["scale"] == "single"
# The control block must not leak into the persisted assistant message.
assert "roboco-meta" not in body["messages"][-1]["content"]
@pytest.mark.asyncio
@@ -500,7 +584,7 @@ async def test_full_happy_path(
headers=_HDR,
)
assert step2b.status_code == HTTPStatus.OK
messages = step2b.json()
messages = step2b.json()["messages"]
assert len(messages) == _DOUBLE_TURN_MSGS
# Step 3: Get draft
@@ -578,7 +662,9 @@ async def test_prompter_chat_draft_ready(prompter_client: dict) -> None:
client = prompter_client["client"]
mock_response = (
"I have enough information. draft_ready=true. Ready to generate a draft."
"Understood.\n\n"
'```roboco-meta\n{"covered": ["objective", "scope", "surface", '
'"acceptance"], "ready": true, "scale": "single"}\n```'
)
with patch(
tests/unit/agent_sdk/test_intake_driver.py
+267
View File
@@ -0,0 +1,267 @@
"""Unit tests for the intake driver loop + event normalization.
SDK-free: the `claude-agent-sdk` message types are stood in by tiny fakes named
the same way `normalize` keys off (`StreamEvent`, `AssistantMessage`, ...), and
the driver loop runs against a fake session/source/sink. The real
`SdkIntakeSession` adapter needs the live `claude` binary and is excluded from
coverage.
"""
from __future__ import annotations
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING
import pytest
from roboco.agent_sdk.intake_driver import (
IntakeDriver,
StreamChunk,
normalize,
)
if TYPE_CHECKING:
from collections.abc import AsyncIterator
# ---------------------------------------------------------------------------
# Fakes mirroring the claude-agent-sdk message/block shapes
# ---------------------------------------------------------------------------
class StreamEvent:
def __init__(self, event: dict) -> None:
self.event = event
class AssistantMessage:
def __init__(self, content: list) -> None:
self.content = content
class ResultMessage:
def __init__(self, session_id: str, total_cost_usd: float | None = None) -> None:
self.session_id = session_id
self.total_cost_usd = total_cost_usd
class SystemMessage:
def __init__(self, subtype: str) -> None:
self.subtype = subtype
class TextBlock:
def __init__(self, text: str) -> None:
self.text = text
class ThinkingBlock:
def __init__(self, thinking: str) -> None:
self.thinking = thinking
class ToolUseBlock:
def __init__(self, name: str, tool_input: dict) -> None:
self.name = name
self.input = tool_input
# ---------------------------------------------------------------------------
# normalize()
# ---------------------------------------------------------------------------
def test_normalize_stream_event_text_delta() -> None:
msg = StreamEvent({"delta": {"type": "text_delta", "text": "hel"}})
chunks = normalize(msg)
assert chunks == [StreamChunk(kind="text", text="hel")]
def test_normalize_stream_event_non_text_delta_is_dropped() -> None:
assert normalize(StreamEvent({"delta": {"type": "input_json_delta"}})) == []
assert normalize(StreamEvent({})) == []
def test_normalize_assistant_message_blocks() -> None:
# Text is NOT re-emitted from the AssistantMessage (the StreamEvent deltas
# already carried it live) — only thinking + tool_use, which have no deltas.
msg = AssistantMessage(
[
TextBlock("hello"),
ThinkingBlock("hmm"),
ToolUseBlock("Read", {"file": "metrics.tsx"}),
]
)
chunks = normalize(msg)
assert [c.kind for c in chunks] == ["thinking", "tool_use"]
assert chunks[0].text == "hmm"
assert chunks[1].tool == "Read"
assert chunks[1].data["input"] == {"file": "metrics.tsx"}
def test_normalize_assistant_message_extracts_draft_block() -> None:
# A finished reply that ends with a fenced roboco-draft block yields a
# single `draft` chunk carrying the parsed object — and no `text` chunk.
text = (
"Here is the task.\n"
"```roboco-draft\n"
'{"title": "Add metrics", "acceptance_criteria": ["x"], "scale": "single"}\n'
"```\n"
)
chunks = normalize(AssistantMessage([TextBlock(text)]))
assert [c.kind for c in chunks] == ["draft"]
assert chunks[0].data["title"] == "Add metrics"
assert chunks[0].data["scale"] == "single"
def test_normalize_assistant_message_malformed_draft_is_ignored() -> None:
bad = "```roboco-draft\n{not valid json}\n```"
assert normalize(AssistantMessage([TextBlock(bad)])) == []
# A draft block with no title is not a usable draft either.
no_title = '```roboco-draft\n{"acceptance_criteria": []}\n```'
assert normalize(AssistantMessage([TextBlock(no_title)])) == []
def test_normalize_propose_draft_tool_becomes_draft_chunk() -> None:
# The canonical signal: the agent CALLS propose_draft → one `draft` chunk
# (not a tool_use chunk).
msg = AssistantMessage(
[
ToolUseBlock(
"propose_draft",
{"draft": {"title": "Add metrics", "acceptance_criteria": ["x"]}},
)
]
)
chunks = normalize(msg)
assert [c.kind for c in chunks] == ["draft"]
assert chunks[0].data["title"] == "Add metrics"
def test_normalize_propose_draft_accepts_flat_input() -> None:
# Tolerant of the draft fields passed flat (no "draft" wrapper).
msg = AssistantMessage([ToolUseBlock("propose_draft", {"title": "Flat", "x": 1})])
chunks = normalize(msg)
assert [c.kind for c in chunks] == ["draft"]
assert chunks[0].data["title"] == "Flat"
def test_normalize_propose_draft_namespaced_name() -> None:
# However the SDK namespaces it (e.g. mcp__intake__propose_draft).
msg = AssistantMessage(
[ToolUseBlock("mcp__intake__propose_draft", {"draft": {"title": "NS"}})]
)
assert [c.kind for c in normalize(msg)] == ["draft"]
def test_normalize_other_tool_stays_tool_use() -> None:
chunks = normalize(AssistantMessage([ToolUseBlock("Read", {"file": "x.py"})]))
assert [c.kind for c in chunks] == ["tool_use"]
assert chunks[0].tool == "Read"
def test_normalize_propose_draft_without_title_is_ignored() -> None:
msg = AssistantMessage(
[ToolUseBlock("propose_draft", {"draft": {"acceptance_criteria": []}})]
)
assert normalize(msg) == []
def test_normalize_result_message_carries_session_id() -> None:
cost = 0.01
chunks = normalize(ResultMessage(session_id="sess-123", total_cost_usd=cost))
assert len(chunks) == 1
assert chunks[0].kind == "turn_end"
assert chunks[0].data["session_id"] == "sess-123"
assert chunks[0].data["cost_usd"] == cost
def test_normalize_system_message() -> None:
chunks = normalize(SystemMessage(subtype="init"))
assert chunks == [StreamChunk(kind="system", data={"subtype": "init"})]
def test_normalize_unknown_message_is_empty() -> None:
assert normalize(object()) == []
# ---------------------------------------------------------------------------
# IntakeDriver loop
# ---------------------------------------------------------------------------
class _FakeSession:
"""Scripts each input text to a list of chunks to stream back."""
def __init__(self, scripted: dict[str, list[StreamChunk]]) -> None:
self.scripted = scripted
self.seen: list[str] = []
async def send(self, text: str) -> AsyncIterator[StreamChunk]:
self.seen.append(text)
for chunk in self.scripted.get(text, []):
yield chunk
class _RaisingSession:
"""Streams one chunk, then fails mid-turn (faithful to a live SDK error)."""
async def send(self, _text: str) -> AsyncIterator[StreamChunk]:
yield StreamChunk(kind="text", text="partial")
raise RuntimeError("boom")
def _source(messages: list[str | None]):
queue = list(messages)
async def _next() -> str | None:
return queue.pop(0) if queue else None
return _next
@pytest.mark.asyncio
async def test_driver_streams_turns_until_shutdown() -> None:
session = _FakeSession(
{
"hi": [StreamChunk(kind="text", text="hello there")],
"more": [
StreamChunk(kind="tool_use", tool="Read"),
StreamChunk(kind="text", text="done"),
],
}
)
@asynccontextmanager
async def factory():
yield session
collected: list[StreamChunk] = []
async def emit(chunk: StreamChunk) -> None:
collected.append(chunk)
driver = IntakeDriver(factory, _source(["hi", "more", None]), emit)
await driver.run()
assert session.seen == ["hi", "more"] # stopped on None, did not call send(None)
assert [c.kind for c in collected] == ["text", "tool_use", "text"]
assert collected[0].text == "hello there"
@pytest.mark.asyncio
async def test_driver_turn_failure_emits_error_and_continues() -> None:
@asynccontextmanager
async def factory():
yield _RaisingSession()
collected: list[StreamChunk] = []
async def emit(chunk: StreamChunk) -> None:
collected.append(chunk)
driver = IntakeDriver(factory, _source(["boom-please", None]), emit)
await driver.run() # must not raise
# The partial chunk made it out, then the failure surfaced as an error chunk.
assert [c.kind for c in collected] == ["text", "error"]
assert collected[0].text == "partial"
assert "boom" in collected[1].text
tests/unit/agent_sdk/test_intake_main.py
+75
View File
@@ -0,0 +1,75 @@
"""Unit tests for the intake container entrypoint wiring helpers."""
from __future__ import annotations
import asyncio
import json
from http import HTTPStatus
import httpx
import pytest
from httpx import ASGITransport, AsyncClient
from roboco.agent_sdk.intake_driver import StreamChunk
from roboco.agent_sdk.intake_main import (
build_receiver,
make_message_source,
make_relay_sink,
)
@pytest.mark.asyncio
async def test_message_source_returns_queued_then_none() -> None:
queue: asyncio.Queue[str | None] = asyncio.Queue()
source = make_message_source(queue)
await queue.put("hi")
await queue.put(None)
assert await source() == "hi"
assert await source() is None # shutdown sentinel
@pytest.mark.asyncio
async def test_relay_sink_posts_chunk_to_orchestrator() -> None:
seen: dict[str, object] = {}
def handler(request: httpx.Request) -> httpx.Response:
seen["url"] = str(request.url)
seen["body"] = json.loads(request.content)
return httpx.Response(200)
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
sink = make_relay_sink("http://orch:8000", "sess-1", client)
await sink(StreamChunk(kind="text", text="hello"))
assert seen["url"] == "http://orch:8000/api/prompter/live/sess-1/events"
assert seen["body"] == {"kind": "text", "text": "hello", "tool": "", "data": {}}
await client.aclose()
@pytest.mark.asyncio
async def test_relay_sink_swallows_post_failure() -> None:
def boom(_request: httpx.Request) -> httpx.Response:
raise httpx.ConnectError("down")
client = httpx.AsyncClient(transport=httpx.MockTransport(boom))
sink = make_relay_sink("http://orch:8000", "sess-1", client)
await sink(StreamChunk(kind="text", text="x")) # must not raise
await client.aclose()
@pytest.mark.asyncio
async def test_receiver_enqueues_turn_and_validates() -> None:
queue: asyncio.Queue[str | None] = asyncio.Queue()
app = build_receiver(queue)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
ok = await client.post("/turn", json={"text": "build a thing"})
assert ok.status_code == HTTPStatus.OK
assert ok.json() == {"queued": True}
assert queue.get_nowait() == "build a thing"
bad = await client.post("/turn", json={"text": ""})
assert bad.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
health = await client.get("/health")
assert health.json() == {"status": "ok"}
tests/unit/api/test_schemas_prompter.py
+76 -16
View File
@@ -10,11 +10,12 @@ from uuid import uuid4
import pytest
from pydantic import ValidationError as PydanticValidationError
from roboco.api.schemas.prompter import (
CellWork,
ChatMessage,
PrompterChatRequest,
PrompterDraftTask,
PrompterMessageRequest,
PrompterSessionCreateRequest,
PrompterTurnResponse,
TaskConfirmRequest,
)
@@ -40,21 +41,6 @@ def test_chat_message_empty_content() -> None:
ChatMessage(role="user", content="")
# =============================================================================
# PrompterSessionCreateRequest
# =============================================================================
def test_session_create_request_defaults() -> None:
req = PrompterSessionCreateRequest()
assert req.context == {}
def test_session_create_request_with_context() -> None:
req = PrompterSessionCreateRequest(context={"team": "backend"})
assert req.context == {"team": "backend"}
# =============================================================================
# PrompterMessageRequest
# =============================================================================
@@ -196,6 +182,80 @@ def test_draft_task_priority_bounds() -> None:
)
# =============================================================================
# Structured spec fields
# =============================================================================
def test_cell_work_valid() -> None:
cw = CellWork(team="backend", summary="Build the endpoint", items=["Route", "Test"])
assert cw.team.value == "backend"
assert cw.items == ["Route", "Test"]
def test_cell_work_requires_summary() -> None:
with pytest.raises(PydanticValidationError):
CellWork(team="backend", summary="")
def test_draft_task_structured_fields_default_empty() -> None:
draft = PrompterDraftTask(
title="Add login page",
description="Implement a secure login page with email and password",
acceptance_criteria=["User can log in"],
team="frontend",
task_type="code",
nature="technical",
estimated_complexity="medium",
)
assert draft.objective is None
assert draft.what_this_builds == []
assert draft.the_work == []
assert draft.notes == []
def test_draft_task_with_structured_fields() -> None:
draft = PrompterDraftTask(
title="Ship the Prompter",
description="A board-led feature spanning three cells, fully wired.",
acceptance_criteria=["It works end to end"],
team="backend",
task_type="code",
nature="technical",
estimated_complexity="high",
objective="Let humans chat a task into existence.",
what_this_builds=["A /prompter page", "A chat endpoint"],
the_work=[
CellWork(team="backend", summary="Chat endpoint", items=["Route"]),
CellWork(team="frontend", summary="Chat UI", items=["Page"]),
],
notes=["Reuse the LLM service"],
)
assert [w.team.value for w in draft.the_work] == ["backend", "frontend"]
def test_confirm_request_carries_edited_draft() -> None:
draft = PrompterDraftTask(
title="Edited title",
description="An edited description that clears the minimum length.",
acceptance_criteria=["Done"],
team="frontend",
task_type="code",
nature="technical",
estimated_complexity="low",
)
req = TaskConfirmRequest(project_id=uuid4(), draft=draft)
assert req.draft is not None
assert req.draft.title == "Edited title"
def test_turn_response_shape() -> None:
resp = PrompterTurnResponse(messages=[], draft_ready=True, scale="multi")
assert resp.draft_ready is True
assert resp.scale == "multi"
assert resp.messages == []
# =============================================================================
# PrompterChatRequest (legacy)
# =============================================================================
tests/unit/runtime/test_budget_sweep.py
+66
View File
@@ -0,0 +1,66 @@
"""``_fetch_budget_status`` — reading an agent's SDK budget endpoint.
Extracted from the budget kill-switch sweep so the swallow of an unreachable
SDK is observable (logged) rather than a silent ``try/except/continue``. These
tests pin the contract: a dict on 200-JSON, ``None`` on every benign failure.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import httpx
import pytest
from roboco.runtime.orchestrator import AgentOrchestrator
if TYPE_CHECKING:
from collections.abc import Callable
_URL = "http://roboco-agent-be-dev-1:9000/budget/status"
def _client(handler: Callable[[httpx.Request], httpx.Response]) -> httpx.AsyncClient:
return httpx.AsyncClient(transport=httpx.MockTransport(handler))
@pytest.mark.asyncio
async def test_returns_dict_on_200_json() -> None:
client = _client(lambda _r: httpx.Response(200, json={"halt": True, "total": 99}))
data = await AgentOrchestrator._fetch_budget_status(client, _URL, "be-dev-1")
assert data == {"halt": True, "total": 99}
await client.aclose()
@pytest.mark.asyncio
async def test_returns_none_when_unreachable() -> None:
def _boom(_r: httpx.Request) -> httpx.Response:
raise httpx.ConnectError("no route to host")
client = _client(_boom)
data = await AgentOrchestrator._fetch_budget_status(client, _URL, "be-dev-1")
assert data is None # benign: container not up yet / gone
await client.aclose()
@pytest.mark.asyncio
async def test_returns_none_on_non_200() -> None:
client = _client(lambda _r: httpx.Response(503))
data = await AgentOrchestrator._fetch_budget_status(client, _URL, "be-dev-1")
assert data is None
await client.aclose()
@pytest.mark.asyncio
async def test_returns_none_on_non_json_body() -> None:
client = _client(lambda _r: httpx.Response(200, text="not json"))
data = await AgentOrchestrator._fetch_budget_status(client, _URL, "be-dev-1")
assert data is None
await client.aclose()
@pytest.mark.asyncio
async def test_returns_none_when_json_is_not_an_object() -> None:
client = _client(lambda _r: httpx.Response(200, json=[1, 2, 3]))
data = await AgentOrchestrator._fetch_budget_status(client, _URL, "be-dev-1")
assert data is None # a list is not a status object
await client.aclose()
tests/unit/runtime/test_intake_spawn.py
+419
View File
@@ -0,0 +1,419 @@
"""The persistent intake (prompter) live-session spawn/reap path.
The intake agent is not task-driven: ``spawn_intake_session`` launches a
long-lived Agent-SDK driver container (image ENTRYPOINT, NOT ``claude -p``),
clones the chat scope's repo(s), and registers the live relay session. These
tests cover the docker-command construction, scope resolution, and the
spawn/reap orchestration with docker + clone mocked (no daemon, no NAS).
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from unittest.mock import patch
from uuid import UUID
import pytest
from roboco.runtime.orchestrator import (
INTAKE_AGENT_ID,
AgentInstance,
AgentOrchestrator,
_IntakeRunSpec,
)
from roboco.services import prompter_live
def _make_minimal_orchestrator() -> AgentOrchestrator:
"""AgentOrchestrator with constructor I/O skipped; _instances ready."""
with patch.object(AgentOrchestrator, "__init__", return_value=None):
orch = AgentOrchestrator.__new__(AgentOrchestrator)
orch._instances = {}
orch._bg_tasks = set()
return orch
def _spec(**overrides: Any) -> _IntakeRunSpec:
base: dict[str, Any] = {
"container_name": "roboco-agent-intake-1",
"image": "roboco-agent-prompter",
"hosts": {
"claude": "/home/runner/.claude",
"prompt": "/data/prompts-generated/intake-1-prompt.md",
"workspaces": "/data/workspaces",
},
"session_id": "sess-abc",
"cwd": "/data/workspaces/roboco/board/intake-1",
"cli_model": "claude-opus-4-6",
"api_url": "http://roboco-orchestrator:8000",
"provider_base_url": None,
"provider_auth_token": None,
}
base.update(overrides)
return _IntakeRunSpec(**base)
@pytest.fixture(autouse=True)
def _fresh_registry() -> Any:
"""Isolate the process-wide live registry per test."""
prev = prompter_live._RegistryHolder.instance
prompter_live._RegistryHolder.instance = prompter_live.PrompterLiveRegistry()
yield
prompter_live._RegistryHolder.instance = prev
# ---------------------------------------------------------------------------
# _build_intake_run_cmd — the pure docker-argv builder.
# ---------------------------------------------------------------------------
class TestBuildIntakeRunCmd:
def test_image_is_last_and_no_claude_cli_args(self) -> None:
cmd = AgentOrchestrator._build_intake_run_cmd(_spec())
assert cmd[-1] == "roboco-agent-prompter"
# The image ENTRYPOINT is the driver — none of the claude CLI flags
# the task-driven path appends may appear here.
for flag in (
"-p",
"--model",
"--system-prompt-file",
"--mcp-config",
"--tools",
):
assert flag not in cmd, f"{flag} must not be in the intake run cmd"
def test_no_workdir_settings_or_manifest_mounts(self) -> None:
cmd = AgentOrchestrator._build_intake_run_cmd(_spec())
joined = " ".join(cmd)
assert "-w" not in cmd # driver sets cwd via ROBOCO_WORKSPACE/the SDK
assert "settings.json" not in joined # no hook mount (driver owns 9000)
assert "mcp-config.json" not in joined # MCP-free live agent
assert "tool-manifest.json" not in joined
def test_env_carries_session_workspace_and_api(self) -> None:
cmd = AgentOrchestrator._build_intake_run_cmd(_spec())
assert "ROBOCO_PROMPTER_SESSION_ID=sess-abc" in cmd
assert "ROBOCO_WORKSPACE=/data/workspaces/roboco/board/intake-1" in cmd
assert "ROBOCO_API_URL=http://roboco-orchestrator:8000" in cmd
assert "ROBOCO_AGENT_ID=intake-1" in cmd
assert "CLAUDE_CODE_SUBAGENT_MODEL=claude-opus-4-6" in cmd
def test_mounts_prompt_and_workspaces(self) -> None:
cmd = AgentOrchestrator._build_intake_run_cmd(_spec())
assert (
"/data/prompts-generated/intake-1-prompt.md:/app/system-prompt.md:ro" in cmd
)
assert "/data/workspaces:/data/workspaces" in cmd
def test_anthropic_default_omits_provider_env(self) -> None:
cmd = AgentOrchestrator._build_intake_run_cmd(_spec())
joined = " ".join(cmd)
assert "ANTHROPIC_BASE_URL" not in joined
assert "ANTHROPIC_AUTH_TOKEN" not in joined
def test_non_anthropic_injects_provider_env(self) -> None:
cmd = AgentOrchestrator._build_intake_run_cmd(
_spec(provider_base_url="http://ollama:11434/v1", provider_auth_token="tok")
)
assert "ANTHROPIC_BASE_URL=http://ollama:11434/v1" in cmd
assert "ANTHROPIC_AUTH_TOKEN=tok" in cmd
# ---------------------------------------------------------------------------
# _intake_scope_slugs — project XOR product resolution.
# ---------------------------------------------------------------------------
class TestIntakeScopeSlugs:
@pytest.mark.asyncio
async def test_project_scope_returns_single_slug(self) -> None:
slugs = await AgentOrchestrator._intake_scope_slugs(
db=object(), project_slug="roboco", product_id=None
)
assert slugs == ["roboco"]
@pytest.mark.asyncio
async def test_product_scope_resolves_distinct_projects_in_order(self) -> None:
# distinct_project_ids returns UUIDs in deterministic team order; the
# primary (cwd) is the first, so order must be preserved (not sorted).
ids = [
"11111111-1111-1111-1111-111111111111",
"22222222-2222-2222-2222-222222222222",
]
class _FakeProduct:
def __init__(self, _db: Any) -> None: ...
async def distinct_project_ids(self, _pid: Any) -> list[Any]:
return [UUID(i) for i in ids]
class _FakeProjectSvc:
async def get(self, pid: Any) -> Any:
return SimpleNamespace(slug=f"proj-{str(pid)[0]}")
with (
patch("roboco.services.product.ProductService", _FakeProduct),
patch(
"roboco.services.project.get_project_service",
lambda _db: _FakeProjectSvc(),
),
):
slugs = await AgentOrchestrator._intake_scope_slugs(
db=object(),
project_slug=None,
product_id="33333333-3333-3333-3333-333333333333",
)
assert slugs == ["proj-1", "proj-2"]
@pytest.mark.asyncio
async def test_product_with_no_projects_raises(self) -> None:
class _FakeProduct:
def __init__(self, _db: Any) -> None: ...
async def distinct_project_ids(self, _pid: Any) -> list[Any]:
return []
with (
patch("roboco.services.product.ProductService", _FakeProduct),
patch("roboco.services.project.get_project_service", lambda _db: object()),
pytest.raises(ValueError, match="no projects"),
):
await AgentOrchestrator._intake_scope_slugs(
db=object(),
project_slug=None,
product_id="33333333-3333-3333-3333-333333333333",
)
# ---------------------------------------------------------------------------
# spawn_intake_session / reap_intake_session — orchestration (docker mocked).
# ---------------------------------------------------------------------------
def _fake_route() -> SimpleNamespace:
return SimpleNamespace(
provider_type=SimpleNamespace(value="anthropic"),
model_name="opus",
base_url=None,
auth_token=None,
)
def _wire_spawn_mocks(
monkeypatch: pytest.MonkeyPatch,
orch: AgentOrchestrator,
run_calls: list[list[str]],
) -> None:
"""Patch every external boundary spawn_intake_session touches."""
async def _clone(_p: Any, _pr: Any) -> tuple[str, list[str]]:
return "/data/workspaces/roboco/board/intake-1", [
"/data/workspaces/roboco/board/intake-1"
]
async def _route(_aid: str) -> Any:
return _fake_route()
async def _noop(*_a: Any, **_k: Any) -> None:
return None
async def _run(cmd: list[str]) -> str:
run_calls.append(cmd)
return "containerid0123456789"
monkeypatch.setattr(orch, "_clone_intake_scope", _clone)
monkeypatch.setattr(orch, "_resolve_agent_route", _route)
monkeypatch.setattr(orch, "_ensure_agent_image", _noop)
monkeypatch.setattr(orch, "_remove_container", _noop)
monkeypatch.setattr(orch, "_run_container_cmd", _run)
monkeypatch.setattr(orch, "_fire_audit", lambda **_k: None)
monkeypatch.setattr(
orch,
"_generate_composed_prompt",
lambda _aid: Path("/tmp/intake-1-prompt.md"),
)
monkeypatch.setattr(
orch,
"_resolve_intake_host_paths",
lambda: {
"claude": "/home/runner/.claude",
"prompt": "/data/prompts-generated/intake-1-prompt.md",
"workspaces": "/data/workspaces",
},
)
class TestSpawnIntakeSession:
@pytest.mark.asyncio
async def test_spawn_registers_session_and_instance(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
orch = _make_minimal_orchestrator()
run_calls: list[list[str]] = []
_wire_spawn_mocks(monkeypatch, orch, run_calls)
instance = await orch.spawn_intake_session("sess-1", project_slug="roboco")
# Live relay session opened for the container.
session = prompter_live.get_live_registry().get("sess-1")
assert session is not None
assert session.agent_id == INTAKE_AGENT_ID
# Orchestrator instance tracked and marked active.
assert orch._instances[INTAKE_AGENT_ID] is instance
assert instance.container_id == "containerid0123456789"
# The cloned cwd reached the docker cmd.
assert "ROBOCO_WORKSPACE=/data/workspaces/roboco/board/intake-1" in run_calls[0]
@pytest.mark.asyncio
async def test_scope_must_be_exactly_one(self) -> None:
orch = _make_minimal_orchestrator()
with pytest.raises(ValueError, match="exactly one"):
await orch.spawn_intake_session("s", project_slug="roboco", product_id="p")
with pytest.raises(ValueError, match="exactly one"):
await orch.spawn_intake_session("s")
@pytest.mark.asyncio
async def test_spawn_reaps_prior_session_first(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
orch = _make_minimal_orchestrator()
run_calls: list[list[str]] = []
_wire_spawn_mocks(monkeypatch, orch, run_calls)
stopped: list[str] = []
async def _stop(aid: str, **_kw: Any) -> None:
stopped.append(aid)
monkeypatch.setattr(orch, "stop_agent", _stop)
# A prior live container already registered for this agent.
orch._instances[INTAKE_AGENT_ID] = AgentInstance(agent_id=INTAKE_AGENT_ID)
await orch.spawn_intake_session("sess-2", project_slug="roboco")
assert stopped == [INTAKE_AGENT_ID] # the old one was reaped first
@pytest.mark.asyncio
async def test_initial_message_is_scheduled(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
orch = _make_minimal_orchestrator()
run_calls: list[list[str]] = []
_wire_spawn_mocks(monkeypatch, orch, run_calls)
scheduled: list[tuple[str, str]] = []
monkeypatch.setattr(
orch,
"_schedule_intake_first_message",
lambda sid, text: scheduled.append((sid, text)),
)
await orch.spawn_intake_session(
"sess-3", project_slug="roboco", initial_message="build X"
)
assert scheduled == [("sess-3", "build X")]
class TestStartIntakeSession:
"""Non-blocking start: relay opens synchronously, spawn runs in the background."""
@pytest.mark.asyncio
async def test_opens_relay_now_and_schedules_spawn(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
orch = _make_minimal_orchestrator()
spawned: list[str] = []
async def _spawn(session_id: str, **_kw: Any) -> Any:
spawned.append(session_id)
return AgentInstance(agent_id=INTAKE_AGENT_ID)
monkeypatch.setattr(orch, "_spawn_intake_container", _spawn)
await orch.start_intake_session("sess-A", project_slug="roboco")
# Relay is open the instant start returns — the SSE stream can connect
# before the (slow) container spawn finishes.
assert prompter_live.get_live_registry().get("sess-A") is not None
await asyncio.sleep(0) # let the scheduled bg spawn run
assert spawned == ["sess-A"]
@pytest.mark.asyncio
async def test_rejects_bad_scope(self) -> None:
orch = _make_minimal_orchestrator()
with pytest.raises(ValueError, match="exactly one"):
await orch.start_intake_session("s", project_slug="r", product_id="p")
class TestSpawnGuarded:
"""A background spawn failure surfaces on the relay instead of dying silently."""
@pytest.mark.asyncio
async def test_failure_pushes_error_and_closes(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
orch = _make_minimal_orchestrator()
registry = prompter_live.get_live_registry()
registry.open("sess-B", INTAKE_AGENT_ID)
pushed: list[tuple[str, dict[str, Any]]] = []
closed: list[str] = []
async def _boom(_session_id: str, **_kw: Any) -> Any:
raise RuntimeError("clone exploded")
def _push(sid: str, ev: dict[str, Any]) -> bool:
pushed.append((sid, ev))
return True
monkeypatch.setattr(orch, "_spawn_intake_container", _boom)
monkeypatch.setattr(registry, "push", _push)
monkeypatch.setattr(registry, "close", closed.append)
await orch._spawn_intake_container_guarded(
"sess-B", project_slug="roboco", product_id=None, initial_message=None
)
assert len(pushed) == 1
assert pushed[0][1]["kind"] == "error"
assert "clone exploded" in pushed[0][1]["text"]
assert closed == ["sess-B"]
class TestReapIntakeSession:
@pytest.mark.asyncio
async def test_reap_closes_session_and_stops_container(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
orch = _make_minimal_orchestrator()
stopped: list[str] = []
async def _stop(aid: str, **_kw: Any) -> None:
stopped.append(aid)
monkeypatch.setattr(orch, "stop_agent", _stop)
registry = prompter_live.get_live_registry()
registry.open("sess-x", INTAKE_AGENT_ID)
await orch.reap_intake_session("sess-x")
assert registry.get("sess-x") is None # relay session closed
assert stopped == [INTAKE_AGENT_ID]
class TestDeliverWhenReady:
@pytest.mark.asyncio
async def test_retries_until_receiver_is_up(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
orch = _make_minimal_orchestrator()
registry = prompter_live.get_live_registry()
succeed_on = 2 # fails once, then succeeds
attempts = {"n": 0}
async def _deliver(_sid: str, _text: str) -> bool:
attempts["n"] += 1
return attempts["n"] >= succeed_on
monkeypatch.setattr(registry, "deliver", _deliver)
await orch._deliver_when_ready("sess-y", "hi", attempts=5, delay=0)
assert attempts["n"] == succeed_on # stopped as soon as delivery succeeded
tests/unit/services/test_prompter.py
+327 -31
View File
@@ -9,19 +9,35 @@ from __future__ import annotations
import json
from typing import Any
from unittest.mock import AsyncMock, patch
from uuid import uuid4
from uuid import UUID, uuid4
import pytest
from roboco.db.tables import AgentTable
from roboco.models.base import AgentRole, AgentStatus
from roboco.db.tables import (
AgentTable,
ProductTable,
ProjectTable,
TaskTable,
)
from roboco.models.base import (
AgentRole,
AgentStatus,
Complexity,
TaskNature,
TaskStatus,
TaskType,
Team,
)
from roboco.seeds.initial_data import AGENT_UUIDS
from roboco.services.base import NotFoundError, ServiceError, ValidationError
from roboco.services.prompter import (
PrompterService,
_build_chat_prompt,
_build_draft_prompt,
_build_reasoning,
_detect_draft_ready,
compose_description,
derive_scale,
get_prompter_service,
parse_readiness,
)
# =============================================================================
@@ -29,27 +45,168 @@ from roboco.services.prompter import (
# =============================================================================
def test_detect_draft_ready_signals() -> None:
signals = [
"I have enough information to proceed",
"Ready to generate a draft now.",
"ready to draft the task",
"i can now draft this.",
"draft_ready=true",
"The task is draft ready",
]
for text in signals:
assert _detect_draft_ready(text), f"Expected True for: {text!r}"
def test_parse_readiness_extracts_and_strips_tag() -> None:
content = (
"Here is my question about scope.\n\n"
'```roboco-meta\n{"covered": ["objective", "scope"], '
'"ready": true, "scale": "multi"}\n```'
)
clean, tag = parse_readiness(content)
assert clean == "Here is my question about scope."
assert tag is not None
assert tag.ready is True
assert tag.scale == "multi"
assert tag.covered == ["objective", "scope"]
# The control block must not leak into the user-visible text.
assert "roboco-meta" not in clean
def test_detect_draft_ready_negative() -> None:
not_signals = [
"Tell me more about the feature.",
"Could you clarify the acceptance criteria?",
"Let's continue the conversation.",
]
for text in not_signals:
assert not _detect_draft_ready(text), f"Expected False for: {text!r}"
def test_parse_readiness_absent_block_is_not_ready() -> None:
clean, tag = parse_readiness("Just a plain reply, no control block.")
assert clean == "Just a plain reply, no control block."
assert tag is None
def test_parse_readiness_malformed_json_is_graceful() -> None:
content = "Reply text.\n```roboco-meta\n{not valid json]\n```"
clean, tag = parse_readiness(content)
assert "roboco-meta" not in clean
assert clean == "Reply text."
assert tag is None
def test_parse_readiness_uses_last_block() -> None:
content = (
'```roboco-meta\n{"ready": false, "scale": "single"}\n```\n'
"Final answer.\n"
'```roboco-meta\n{"ready": true, "scale": "multi"}\n```'
)
clean, tag = parse_readiness(content)
assert tag is not None
assert tag.ready is True
assert tag.scale == "multi"
assert "roboco-meta" not in clean
def test_derive_scale_single_vs_multi() -> None:
assert derive_scale([{"team": "backend"}]) == "single"
assert derive_scale([{"team": "backend"}, {"team": "frontend"}]) == "multi"
# Non-cell teams (e.g. main_pm) do not count toward cell breadth.
assert derive_scale([{"team": "backend"}, {"team": "main_pm"}]) == "single"
assert derive_scale([]) == "single"
def test_compose_description_single_cell_markdown() -> None:
draft = {
"objective": "Let humans track token usage.",
"what_this_builds": ["A usage panel on the Metrics page"],
"the_work": [
{
"team": "frontend",
"summary": "Render the usage panel",
"items": ["Add the chart", "Wire the API"],
}
],
"notes": ["Reuse the existing Metrics layout"],
"acceptance_criteria": ["Panel shows totals", "Panel filters by range"],
}
md = compose_description(draft)
assert "## Objective" in md
assert "## What This Builds" in md
assert "## The Work" in md
assert "**Frontend** — Render the usage panel" in md
assert "## Notes" in md
assert "## Success Criteria" in md
assert "- Panel shows totals" in md
# Single-cell tasks get no board-led lead line.
assert "Board-led" not in md
def test_compose_description_multi_cell_has_board_led_lead() -> None:
draft = {
"objective": "Ship the Prompter.",
"the_work": [
{"team": "backend", "summary": "Chat endpoint", "items": []},
{"team": "frontend", "summary": "Chat UI", "items": []},
{"team": "ux_ui", "summary": "Interaction design", "items": []},
],
"acceptance_criteria": ["It works end to end"],
}
md = compose_description(draft)
assert "Board-led" in md
assert "**Backend**" in md
assert "**UX/UI**" in md
def test_compose_description_falls_back_to_provided_description() -> None:
# Sparse structured fields → fall back to a model-provided description.
draft = {"description": "A perfectly adequate fallback description here."}
md = compose_description(draft)
assert md == "A perfectly adequate fallback description here."
def test_lead_cell_team_prefers_the_work_cell() -> None:
draft = {"the_work": [{"team": "frontend"}], "team": "backend"}
assert PrompterService._lead_cell_team(draft, default=Team.BACKEND) is Team.FRONTEND
# Empty the_work falls back to the provided default.
assert PrompterService._lead_cell_team({}, default=Team.BACKEND) is Team.BACKEND
def test_lead_cell_team_skips_invalid_cell_names() -> None:
# An off-enum cell name is skipped, not raised on; falls through to a valid one.
draft = {"the_work": [{"team": "nonsense"}, {"team": "frontend"}]}
assert PrompterService._lead_cell_team(draft, default=Team.BACKEND) is Team.FRONTEND
def test_coerce_draft_enums_defaults_invalid_values() -> None:
# Regression: the LLM emits off-enum values (e.g. task_type="feature"). The
# confirm must coerce to defaults, never raise — a bad enum guess must not
# 400 the launch and force the agent to self-correct in-chat.
draft = {
"team": "backend",
"task_type": "feature", # not a valid TaskType
"nature": "bogus", # not a valid TaskNature
"estimated_complexity": "enormous", # not a valid Complexity
}
team, task_type, nature, complexity = PrompterService._coerce_draft_enums(draft)
assert team is Team.BACKEND
assert task_type is TaskType.CODE
assert nature is TaskNature.TECHNICAL
assert complexity is Complexity.MEDIUM
def test_coerce_priority_maps_words_clamps_and_defaults() -> None:
# Regression: priority is the one non-enum field the agent guesses, and it
# guesses a word ("high") as often as a number — int("high") used to 500.
# word/number -> expected priority int (0=urgent .. 3=low).
cases: dict[object, int] = {
"urgent": 0,
"high": 1,
"medium": 2,
"low": 3,
1: 1,
"3": 3,
99: 3, # clamped into range
"nonsense": 2, # unrecognized -> default medium
None: 2, # missing -> default medium
}
for value, expected in cases.items():
assert PrompterService._coerce_priority(value) == expected
def test_coerce_draft_enums_keeps_valid_and_derives_missing_team() -> None:
# Valid values pass through; a missing team is derived from the_work.
draft = {
"task_type": "documentation",
"nature": "technical",
"estimated_complexity": "medium",
"the_work": [{"team": "frontend"}],
}
team, task_type, nature, complexity = PrompterService._coerce_draft_enums(draft)
assert team is Team.FRONTEND
assert task_type is TaskType.DOCUMENTATION
assert nature is TaskNature.TECHNICAL
assert complexity is Complexity.MEDIUM
def test_build_chat_prompt_basic() -> None:
@@ -131,17 +288,26 @@ async def test_chat_success_with_mock_llm() -> None:
async def test_chat_draft_ready_signal() -> None:
service = get_prompter_service()
reply = (
"Got it — I have what I need.\n\n"
'```roboco-meta\n{"covered": ["objective", "scope", "surface", '
'"acceptance"], "ready": true, "scale": "single"}\n```'
)
with patch.object(
service,
"_create_message",
new_callable=AsyncMock,
return_value="I have enough information. Ready to draft.",
return_value=reply,
):
result = await service.chat(
messages=[{"role": "user", "content": "I need a feature"}]
)
assert result["draft_ready"] is True
assert result["scale"] == "single"
# The control block is stripped from the user-visible reply.
assert "roboco-meta" not in result["message"]
assert result["message"] == "Got it — I have what I need."
@pytest.mark.asyncio
@@ -245,8 +411,9 @@ async def test_create_session_db(db_session: Any) -> None:
"""create_session persists a PrompterSessionTable row."""
service = get_prompter_service(db=db_session)
agent_id = uuid4()
agent = AgentTable(
id=uuid4(),
id=agent_id,
name="TestAgent",
slug=f"test-{uuid4().hex[:8]}",
role=AgentRole.DEVELOPER,
@@ -261,10 +428,10 @@ async def test_create_session_db(db_session: Any) -> None:
db_session.add(agent)
await db_session.flush()
session = await service.create_session(agent_id=agent.id) # type: ignore[arg-type]
session = await service.create_session(agent_id=agent_id)
assert session.id is not None
assert session.status == "active"
assert session.agent_id == agent.id
assert session.agent_id == agent_id
@pytest.mark.asyncio
@@ -280,8 +447,9 @@ async def test_get_draft_empty_session_raises(db_session: Any) -> None:
"""get_or_generate_draft raises ValidationError if no messages exist."""
service = get_prompter_service(db=db_session)
agent_id = uuid4()
agent = AgentTable(
id=uuid4(),
id=agent_id,
name="TestAgent",
slug=f"test-{uuid4().hex[:8]}",
role=AgentRole.DEVELOPER,
@@ -296,10 +464,138 @@ async def test_get_draft_empty_session_raises(db_session: Any) -> None:
db_session.add(agent)
await db_session.flush()
session = await service.create_session(agent_id=agent.id) # type: ignore[arg-type]
session = await service.create_session(agent_id=agent_id)
with pytest.raises(ValidationError, match="empty conversation"):
await service.get_or_generate_draft(
session_id=session.id, # type: ignore[arg-type]
agent_id=agent.id, # type: ignore[arg-type]
session_id=UUID(str(session.id)),
agent_id=agent_id,
)
async def _seed_project_and_ceo(db_session: Any) -> tuple[UUID, UUID]:
"""Seed a system agent + project + CEO; return (project_id, ceo_id).
Returns plain ``UUID``s (not the ORM rows) so callers pass real uuids to the
service — no casting the ORM ``.id`` column type at the call site.
"""
system_id, project_id, ceo_id = uuid4(), uuid4(), uuid4()
system = AgentTable(
id=system_id,
name="System",
slug=f"system-{uuid4().hex[:8]}",
role=AgentRole.SYSTEM,
team=None,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="system",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(system)
await db_session.flush()
project = ProjectTable(
id=project_id,
name="Intake Test Project",
slug=f"intake-{uuid4().hex[:8]}",
git_url="https://github.com/example/intake.git",
default_branch="main",
protected_branches=["main"],
assigned_cell=Team.BACKEND,
created_by=system_id,
is_active=True,
)
ceo = AgentTable(
id=ceo_id,
name="CEO",
slug=f"ceo-{uuid4().hex[:8]}",
role=AgentRole.CEO,
team=None,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="ceo",
capabilities=[],
permissions={},
metrics={},
)
db_session.add_all([project, ceo])
await db_session.flush()
return project_id, ceo_id
@pytest.mark.asyncio
async def test_confirm_live_draft_board_route_assigns_po(db_session: Any) -> None:
""" "Board review & Start" (default route) → PENDING, assigned to the Product
Owner so the orchestrator fires the PO + HoM review."""
project_id, ceo_id = await _seed_project_and_ceo(db_session)
service = get_prompter_service(db=db_session)
draft = {
"title": "Add token metrics",
"objective": "See token usage at a glance.",
"acceptance_criteria": ["Dashboard shows total tokens"],
"team": "backend",
"the_work": [
{"team": "backend", "summary": "instrument", "items": ["count tokens"]}
],
}
task_id = await service.confirm_live_draft(draft, ceo_id, project_id=project_id)
row = await db_session.get(TaskTable, task_id)
assert row is not None
assert row.status == TaskStatus.PENDING # "& Start" — started now
assert row.assigned_to == UUID(AGENT_UUIDS["product-owner"]) # board review
assert row.source == "prompter"
assert row.confirmed_by_human is True
assert row.team == Team.BACKEND # lead cell from the_work
assert row.created_by == ceo_id
assert row.nature is not None and row.task_type is not None
@pytest.mark.asyncio
async def test_confirm_live_draft_main_pm_route_assigns_main_pm(
db_session: Any,
) -> None:
""" "Approve & Start" (route="main_pm") → PENDING, assigned to the Main PM."""
project_id, ceo_id = await _seed_project_and_ceo(db_session)
service = get_prompter_service(db=db_session)
draft = {
"title": "Quick fix",
"acceptance_criteria": ["done"],
"team": "backend",
}
task_id = await service.confirm_live_draft(
draft, ceo_id, project_id=project_id, route="main_pm"
)
row = await db_session.get(TaskTable, task_id)
assert row.status == TaskStatus.PENDING
assert row.assigned_to == UUID(AGENT_UUIDS["main-pm"])
@pytest.mark.asyncio
async def test_confirm_live_draft_product_routes_to_main_pm(db_session: Any) -> None:
"""A product-scoped live draft is a board-led coordination root (Main PM)."""
_project_id, ceo_id = await _seed_project_and_ceo(db_session)
product_id = uuid4()
product = ProductTable(
id=product_id,
name="Intake Product",
slug=f"prod-{uuid4().hex[:8]}",
description="x",
created_by=ceo_id,
)
db_session.add(product)
await db_session.flush()
service = get_prompter_service(db=db_session)
draft = {
"title": "Board-led feature",
"acceptance_criteria": ["works end to end"],
"team": "backend",
}
task_id = await service.confirm_live_draft(draft, ceo_id, product_id=product_id)
row = await db_session.get(TaskTable, task_id)
assert row.team == Team.MAIN_PM
assert row.product_id == product_id
assert row.project_id is None
tests/unit/services/test_prompter_live.py
+117
View File
@@ -0,0 +1,117 @@
"""Unit tests for the live intake-session relay (orchestrator side)."""
from __future__ import annotations
import asyncio
import json
import httpx
import pytest
from roboco.services.prompter_live import (
PrompterLiveRegistry,
get_live_registry,
)
def test_open_get_close() -> None:
reg = PrompterLiveRegistry()
session = reg.open("s1", "intake-1")
assert session.agent_id == "intake-1"
assert reg.get("s1") is session
reg.close("s1")
assert reg.get("s1") is None
def test_open_is_idempotent_for_a_live_session() -> None:
"""Re-opening a live session returns the SAME object (same queue).
Regression: a second open() that swapped in a fresh queue orphaned the SSE
stream — the panel had already captured the first queue, so the agent's
replies (pushed to the new queue) never reached the browser.
"""
reg = PrompterLiveRegistry()
first = reg.open("s1", "intake-1")
first.queue.put_nowait({"event": "text"}) # something already queued
second = reg.open("s1", "intake-1")
assert second is first # not replaced
assert second.queue is first.queue # same queue → stream not orphaned
# After close, a re-open starts fresh (no stale queue carried over).
reg.close("s1")
third = reg.open("s1", "intake-1")
assert third is not first
assert third.queue.empty()
def test_push_to_unknown_or_closed_returns_false() -> None:
reg = PrompterLiveRegistry()
assert reg.push("nope", {"event": "text"}) is False
reg.open("s1", "intake-1")
assert reg.push("s1", {"event": "text"}) is True
reg.close("s1")
assert reg.push("s1", {"event": "text"}) is False
@pytest.mark.asyncio
async def test_stream_yields_queued_events_then_ends_on_close() -> None:
reg = PrompterLiveRegistry()
reg.open("s1", "intake-1")
async def collect() -> list[dict]:
return [ev async for ev in reg.stream("s1")]
task = asyncio.create_task(collect())
await asyncio.sleep(0) # let the stream capture the session + block on get()
reg.push("s1", {"event": "text", "data": "hel"})
reg.push("s1", {"event": "turn_end", "data": "{}"})
reg.close("s1") # sentinel ends the stream
result = await asyncio.wait_for(task, timeout=1.0)
assert result == [
{"event": "text", "data": "hel"},
{"event": "turn_end", "data": "{}"},
]
@pytest.mark.asyncio
async def test_stream_unknown_session_is_empty() -> None:
reg = PrompterLiveRegistry()
assert [ev async for ev in reg.stream("nope")] == []
@pytest.mark.asyncio
async def test_deliver_posts_to_the_container_receiver() -> None:
seen: dict[str, object] = {}
def handler(request: httpx.Request) -> httpx.Response:
seen["host"] = request.url.host
seen["path"] = request.url.path
seen["body"] = json.loads(request.content)
return httpx.Response(200, json={"ok": True})
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
reg = PrompterLiveRegistry(http_client=client)
reg.open("s1", "intake-1")
assert await reg.deliver("s1", "hello there") is True
assert seen["host"] == "roboco-agent-intake-1"
assert seen["path"] == "/turn"
assert seen["body"] == {"text": "hello there"}
await client.aclose()
@pytest.mark.asyncio
async def test_deliver_to_unknown_or_failing_returns_false() -> None:
def fail(_request: httpx.Request) -> httpx.Response:
return httpx.Response(500)
client = httpx.AsyncClient(transport=httpx.MockTransport(fail))
reg = PrompterLiveRegistry(http_client=client)
assert await reg.deliver("nope", "hi") is False # unknown session
reg.open("s1", "intake-1")
assert await reg.deliver("s1", "hi") is False # 500 from container
await client.aclose()
def test_registry_singleton() -> None:
assert get_live_registry() is get_live_registry()