Skip to main content

Agentic Runtime Streaming Implementation Plan

For agentic workers: REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Implement #248 token-by-token SSE streaming, then verify and close actual remaining #172 gaps on current dev.

Architecture: Add a streaming subprocess path to the Hermes client, make FastAPI stream token frames plus final normalized metadata, keep NestJS as a raw SSE proxy, and treat #172 as a current-state verification pass against the merged Hermes task runtime.

Tech Stack: Python 3.11+/FastAPI, Hermes CLI subprocesses, pytest, NestJS/TypeScript only if proxy tests need updates, Next.js only if #172 verification finds UI gaps.

File Map​

  • Modify: agent/hermes_client.py
    Add HermesStreamEvent, shared command construction, and chat_stream().
  • Modify: agent/main.py
    Replace single-event _stream_chat_sse() with token forwarding and finalization.
  • Modify: agent/routers/chat.py
    Delegate /v1/chat/stream to the canonical stream implementation.
  • Modify: agent/evals/test_hermes_client_subprocess.py
    Add streaming subprocess tests.
  • Modify: agent/evals/test_endpoints_and_branches.py
    Add SSE token-before-done endpoint tests and update old done-event assertions.
  • Inspect, maybe modify: agent/task_runtime.py, agent/models/tasks.py, api/src/agentic/*, frontend/src/app/workspace/tasks/page.tsx, frontend/src/app/workspace/queue/ExpertWorkspacePane.tsx
    Verify #172 checklist and patch only real gaps.
  • Create/modify docs only if implementation changes public contracts.

Task 1: Hermes Streaming Client​

Files:

  • Modify: agent/hermes_client.py

  • Test: agent/evals/test_hermes_client_subprocess.py

  • Step 1: Write failing test for structured message deltas

@pytest.mark.asyncio
async def test_chat_stream_yields_message_delta_tokens_and_final_result(tmp_path):
    fake_proc = _make_fake_stream_proc([
        b'{"type":"message.delta","delta":"Hello"}\n',
        b'{"type":"message.delta","delta":" there"}\n',
        b'{"type":"message.done"}\n',
        b'session_id: sess_stream\n',
    ])
    ...
    assert [event.event for event in events] == ["token", "token", "done"]
    assert events[-1].result.stdout == "Hello there\nsession_id: sess_stream\n"
  • Step 2: Run red test

Run:

cd agent
./.venv/bin/pytest evals/test_hermes_client_subprocess.py::test_chat_stream_yields_message_delta_tokens_and_final_result -q

Expected: fail because HermesStreamEvent / chat_stream does not exist.

  • Step 3: Implement minimal streaming client

Add:

@dataclass(frozen=True)
class HermesStreamEvent:
    event: str
    token: Optional[str] = None
    result: Optional[HermesRunResult] = None

Then add HermesClient.chat_stream() using asyncio.create_subprocess_exec, stdout chunk reading, structured delta parsing, timeout handling, stderr capture, and a final HermesRunResult.

  • Step 4: Run green test

Run the same pytest command. Expected: pass.

  • Step 5: Add fallback/token parser coverage

Add tests for plain text stdout and timeout. Run:

cd agent
./.venv/bin/pytest evals/test_hermes_client_subprocess.py -q

Expected: pass.

Task 2: Canonical /chat/stream​

Files:

  • Modify: agent/main.py

  • Test: agent/evals/test_endpoints_and_branches.py

  • Step 1: Write failing endpoint test

Add a test that patches main.hermes_client.chat_stream() to yield two token events then a done event with a parseable HermesRunResult.

Expected SSE contains:

event: token
data: {"token": "Hello"}

event: done
data: {"reply": "Hello there", ...}
  • Step 2: Run red test
cd agent
./.venv/bin/pytest evals/test_endpoints_and_branches.py::test_chat_stream_emits_token_events_before_done_event -q

Expected: fail because endpoint still emits a single data event.

  • Step 3: Implement streaming finalization in main.py

Update _stream_chat_sse() to:

  • run the same cost-cap and workspace setup as chat()

  • call hermes_client.chat_stream()

  • yield event: token frames for token events

  • parse the final HermesRunResult

  • normalize reply, confidence, risk_level, flagForReview, session_id, artifacts, and audit

  • yield event: done

  • always terminate with data: [DONE]

  • Step 4: Run endpoint tests

cd agent
./.venv/bin/pytest evals/test_endpoints_and_branches.py::test_chat_stream_emits_token_events_before_done_event evals/test_endpoints_and_branches.py::test_chat_stream_emits_done_event_with_final_payload -q

Expected: pass.

Task 3: /v1/chat/stream Delegation​

Files:

  • Modify: agent/routers/chat.py

  • Test: add or extend agent/evals/test_endpoints_and_branches.py

  • Step 1: Write failing v1 stream test

Post to /v1/chat/stream, patch main.hermes_client.chat_stream(), and assert token frames appear before done.

  • Step 2: Run red test
cd agent
./.venv/bin/pytest evals/test_endpoints_and_branches.py::test_v1_chat_stream_emits_token_events -q

Expected: fail because v1 route buffers through generate_draft_reply().

  • Step 3: Delegate v1 stream to canonical stream

In agent/routers/chat.py, call:

legacy_req = _to_legacy_request(req)
return StreamingResponse(main._stream_chat_sse(legacy_req), ...)
  • Step 4: Run v1 stream test

Expected: pass.

Task 4: NestJS Proxy Check​

Files:

  • Inspect: api/src/common/agent.client.ts

  • Inspect: api/src/conversations/conversations.service.ts

  • Step 1: Verify proxy does not buffer

Confirm AgentClient.chatStream() returns response.body and ConversationsService.streamMessage() writes chunks directly.

  • Step 2: Patch comments/tests only if stale

No code change if proxy is already transparent.

  • Step 3: Run API test if patched
cd api
npm test -- --runInBand path/to/focused.test.ts

Expected: pass, only if a focused API test exists or is added.

Task 5: #172 Current-State Verification​

Files:

  • Inspect: agent/task_runtime.py

  • Inspect: agent/models/tasks.py

  • Inspect: agent/routers/tasks.py

  • Inspect: api/src/agentic/agentic.controller.ts

  • Inspect: api/src/agentic/agentic.service.ts

  • Inspect: frontend/src/app/workspace/tasks/page.tsx

  • Inspect: frontend/src/app/workspace/queue/ExpertWorkspacePane.tsx

  • Step 1: Map issue checklist to current files

Produce a short local checklist:

real execution: file/line evidence
checkpointing: file/line evidence
step approval resume: file/line evidence
failure halt/events: file/line evidence
tool result persistence: file/line evidence
API endpoints/events: file/line evidence
frontend task detail/audit: file/line evidence
  • Step 2: Patch only real gaps

Likely patches, if needed:

  • add missing task history/audit rendering from checkpoint fields

  • add focused tests for current-step resume or failure halt

  • update stale docs that still imply execute_node stubs

  • Step 3: Run focused tests

cd agent
./.venv/bin/pytest tests evals/test_task_runtime*.py -q

Adjust exact file names to existing tests found during inspection.

Task 6: Verification and Gate Recording​

  • Step 1: Run focused agent tests
cd agent
./.venv/bin/pytest evals/test_hermes_client_subprocess.py evals/test_endpoints_and_branches.py -q
  • Step 2: Run broader relevant tests
cd agent
./.venv/bin/pytest tests/ evals/test_hermes_client_subprocess.py evals/test_endpoints_and_branches.py -q
  • Step 3: Record no-bs evidence
python3 ~/.codex/no-bs/bin/no_bs.py record-file <changed-files>
python3 ~/.codex/no-bs/bin/no_bs.py record-verification '<commands and results>'
python3 ~/.codex/no-bs/bin/no_bs.py complete-task '#248 SSE token-by-token streaming'
python3 ~/.codex/no-bs/bin/no_bs.py complete-task '#172 agentic execute_node checkpointing step-level approval'
python3 ~/.codex/no-bs/bin/no_bs.py check

Expected: can_claim_completion true before making a final completion claim.

Last updated on by ethumanity