feat(agentic): ACP turn-continuation — the human-sim decides every turn (reply / nudge / conclude)

.github/workflows/ci.yml

@@ -51,7 +51,7 @@ jobs:
         run: uv run python -m pytest -q -W error::DeprecationWarning
       - name: agentic grading core (hermetic)
         working-directory: agentic/harness
-        run: uv run --project ../.. python -m pytest test_invariants.py test_pool_and_cluster_ops.py test_scenario_knobs.py test_human_sim.py test_runner.py test_provenance.py test_targets.py test_hermes_trace.py test_hermes_setup.py test_acp_client.py -q -W error::DeprecationWarning
+        run: uv run --project ../.. python -m pytest test_invariants.py test_pool_and_cluster_ops.py test_scenario_knobs.py test_human_sim.py test_runner.py test_provenance.py test_targets.py test_hermes_trace.py test_hermes_setup.py test_acp_client.py test_hermes_runner.py -q -W error::DeprecationWarning
 
   audit:
     name: advisories in the lock

agentic/harness/acp_client.py

@@ -149,6 +149,23 @@ def hpc_bridge_mcp(repo_root: str, env: dict[str, str]) -> McpServerStdio:
     )
 
 
+def _join_chunks(chunks: list[str]) -> str:
+    """Reassemble a turn's AgentMessageChunk texts. Over a STREAMING provider (the Argo tunnel) the chunks are token
+    DELTAS — joining them with ' ' put spaces inside words ("part ition", "sp ending"; sonnet-5 via Argo, 2026-09-08)
+    and the human-sim read that garble as the operator's ask. Deltas concatenate verbatim. Over a non-streaming
+    provider (ALCF) a chunk is a whole message; when one ends a sentence and the next starts a new one with no
+    whitespace between them, a newline keeps them apart. (Grading never reads this text — the graded question comes
+    from state.db post-run — but the sim's judgement does.)"""
+    out = ""
+    for c in chunks:
+        if not c:
+            continue
+        if out and out.rstrip()[-1:] in ".!?" and not out[-1].isspace() and c[0].isalpha() and c[0].isupper():
+            out += "\n"
+        out += c
+    return out
+
+
 @dataclass
 class AcpTurn:
     """One turn of an ACP session: the operator's closing prose for that turn (used to decide whether it asked
@@ -180,7 +197,7 @@ async def run_session(command: str, args: list[str], task: str, *, cwd: str, env
             resp = await conn.prompt(prompt=[acp.text_block(prompt_text)], session_id=sess.session_id)
             if respond is None:
                 break
-            turn_text = " ".join(client.capture.texts[seen:]).strip()
+            turn_text = _join_chunks(client.capture.texts[seen:]).strip()
             reply = await respond(AcpTurn(text=turn_text, calls_so_far=len(client.capture.tool_calls)))
             if not reply:
                 break
@@ -204,7 +221,7 @@ async def _probe() -> int:
                                mcp_servers=[hpc_bridge_mcp(repo, env)])
     print(f"stop_reason: {getattr(resp, 'stop_reason', None)}", file=sys.stderr)
     print(f"tool calls: {[c.get('title') or (c.get('raw_input') or {}) for c in cap.tool_calls]}", file=sys.stderr)
-    print("AGENT SAID:", (" ".join(cap.texts)).strip()[-400:])
+    print("AGENT SAID:", _join_chunks(cap.texts).strip()[-400:])
     return 0
 
 

agentic/harness/hermes_runner.py

@@ -18,6 +18,11 @@
 
 The trace comes from hermes' own ``state.db`` (see hermes_trace); prose Q&A is stamped in as synthetic
 AskUserQuestion calls so the interactive graders apply unchanged. Mid-run chaos hooks are not supported here.
+
+The ACP path (``HPCB_HERMES_ACP=1``, ``_run_acp``) is the benchmark driver: ONE persistent session, and after every
+operator turn the human-sim decides reply / nudge / conclude (``HumanSim.move`` via ``AcpResponder``) — a decisive
+operator that ends a turn on a plan gets a "carry on", a wrap-up ends the session, a standing decline is never
+nudged. Nudges are stamped as ``user_nudge`` markers, never as questions.
 """
 from __future__ import annotations
 
@@ -31,7 +36,8 @@
 
 import hermes_setup
 from hermes_trace import exchanges_from_messages, load_messages, stamp_exchanges, trace_from_messages
-from runner import HPC_BRIDGE_TOOLS, MAX_PROSE_FOLLOWUPS, RunResult  # shared dataclass + logical tool names + cap
+from human_sim import MAX_NUDGES, MAX_PROSE_FOLLOWUPS
+from runner import HPC_BRIDGE_TOOLS, RunResult  # shared dataclass + logical tool names
 
 _HERMES_BIN = "hermes"
 # Harness plumbing + Anthropic auth the hermes child has no business seeing. ALCF_INFERENCE_TOKEN is KEPT (hermes
@@ -132,6 +138,30 @@ async def _run_turn(full_prompt: str, child_env: dict[str, str], timeout_s: int)
         raise
 
 
+class AcpResponder:
+    """The ACP session's interactive `respond` hook: after EVERY operator turn, ask the persona'd human-sim what a
+    person does next (`HumanSim.move` — reply / nudge / conclude) and send that, keeping the replies IN ORDER for the
+    post-run stamping. `exchanges_from_messages` pairs each `user` row after the first with `replies[N]`, and a nudge
+    is a user row like any answer, so nudges are recorded too (kind "nudge" → stamped as `user_nudge`, never as a
+    question). Turn CONTINUATION no longer hinges on `ends_with_question`; the guards (a standing decline is never
+    nudged; each budget ends in conclude) live in the sim so they are tested there."""
+
+    def __init__(self, human, persona: str | None) -> None:
+        self.human = human
+        self.persona = persona
+        self.replies: list[dict] = []   # {answer, kind} per message SENT, in order
+
+    async def __call__(self, turn) -> str | None:
+        move = await self.human.move(turn.text)
+        tag = f"{move.kind}{f': {move.reason}' if move.reason else ''}"
+        if move.action == "conclude":
+            print(f"  human({self.persona}) concludes [{tag}]", file=sys.stderr, flush=True)
+            return None
+        self.replies.append({"answer": move.reply, "kind": move.kind})
+        print(f"  human({self.persona}) [{tag}]: {move.reply[:140]}", file=sys.stderr, flush=True)
+        return move.reply
+
+
 def _turn_final_text(rows: list[dict], after_id: int) -> str:
     """The operator's last PROSE (assistant text) in the messages added since ``after_id`` — this turn's closing
     message, which a prose question ends with. Empty when the turn closed on tool calls (task done, no question)."""
@@ -162,38 +192,28 @@ async def _run_acp(prompt: str, *, home: Path, child_env: dict[str, str], alcf_m
     full_prompt = _lead(ablate_skill, interactive=interactive) + "\n\n" + prompt
     db = home / "state.db"
     human = None
-    replies: list[dict] = []   # {answer, kind} per human-sim turn, in order — correlated to prose questions POST-RUN
-    state = {"capped": False}
     respond = None
     if interactive:
-        from human_sim import HumanSim, ends_with_question
+        from human_sim import HumanSim
 
+        # The sim decides EVERY turn (reply / nudge / conclude) off the ACP capture's turn text — reliable and
+        # complete when prompt() returns. Only {answer, kind} is recorded per message sent; the question TEXT + its
+        # trace INDEX are reconstructed POST-RUN from the flushed state.db (exchanges_from_messages) — the ACP
+        # capture's count/joined-chunks misaligned both.
         human = HumanSim(persona=persona, goal=user_goal, totp_secret=os.environ.get("HPCB_SIM_TOTP_SECRET") or None)
-
-        async def respond(turn):
-            # ends_with_question(turn.text) uses the ACP capture (reliable, complete when prompt() returns) to
-            # decide whether the operator asked — this drives turn CONTINUATION and completes runs. We record only
-            # {answer, kind}; the question TEXT + its trace INDEX are reconstructed POST-RUN from the flushed
-            # state.db (exchanges_from_messages) — the ACP capture's count/joined-chunks misaligned both.
-            if not ends_with_question(turn.text):
-                return None        # the operator finished / didn't ask — end the session
-            if len(replies) >= MAX_PROSE_FOLLOWUPS:
-                state["capped"] = True
-                return None
-            reply, kind, reason = await human.reply_hermes(turn.text)
-            replies.append({"answer": reply, "kind": kind})
-            print(f"  human({persona}) [{kind}{f': {reason}' if reason else ''}]: {reply[:140]}",
-                  file=sys.stderr, flush=True)
-            return reply
+        respond = AcpResponder(human, persona)
 
     print(f"  hermes-ACP: model={alcf_model} home={home} "
           f"({'interactive ' + str(persona) if interactive else 'autonomous'})", file=sys.stderr, flush=True)
     t0 = time.time()
     err = False
     stop = "?"
     try:
+        # turns: the task + every answer + every nudge (each budget ends in a `conclude`, so this bound is never hit
+        # by the sim itself — it only guards a respond hook that keeps returning text)
         resp, _cap = await acp_client.run_session("hermes", ["acp"], full_prompt, cwd=repo, env=child_env,
-                                                  mcp_servers=mcp, respond=respond, max_turns=MAX_PROSE_FOLLOWUPS + 1)
+                                                  mcp_servers=mcp, respond=respond,
+                                                  max_turns=1 + MAX_PROSE_FOLLOWUPS + MAX_NUDGES)
         stop = str(getattr(resp, "stop_reason", "") or "")
     except asyncio.CancelledError:
         raise
@@ -206,14 +226,16 @@ async def respond(turn):
     trace = trace_from_messages(rows)
     if interactive:
         # Stamp POST-RUN from the flushed state.db: correlate each recorded reply to the operator's prose question
-        # by message order, so the synthetic AskUserQuestion lands at the right trace index with the clean ask text.
-        trace = stamp_exchanges(trace, exchanges_from_messages(rows, replies))
+        # by message order, so the synthetic AskUserQuestion lands at the right trace index with the clean ask text
+        # (a nudge lands as a `user_nudge` marker instead — it answered no question).
+        trace = stamp_exchanges(trace, exchanges_from_messages(rows, respond.replies))
     answer = trace.texts[-1] if trace.texts else ""
     return RunResult(
         trace=trace,
         final=HermesFinal(result=answer, is_error=err, session_id=(rows[0].get("session_id") if rows else None)),
         messages=rows, dialogue=(human.dialogue if human else []),
-        prose_followups=len(replies), followups_capped=state["capped"],
+        prose_followups=(human.answers if human else 0), followups_capped=(human.followups_capped if human else False),
+        nudges=(human.nudges if human else 0), nudges_capped=(human.nudges_capped if human else False),
         human_sim_model=(human.model if human else None),
     )
 

agentic/harness/hermes_trace.py

@@ -162,18 +162,27 @@ def stamp_exchanges(trace: Trace, exchanges: list[dict] | None) -> Trace:
     interactive graders (spend_follows_question / choice_respected / refusal_exercised / no_spend_after_decline
     — all keyed on ``t.named("AskUserQuestion")`` with the answer in ``ToolCall.answers``) work for the hermes
     operator exactly as for Claude. hermes has no AskUserQuestion tool: the operator asks in prose and the
-    human-sim replies in prose. Each exchange = ``{call_index, question, answer}``; the synthetic call is
+    human-sim replies in prose. Each exchange = ``{call_index, question, answer, kind}``; the synthetic call is
     inserted right AFTER ``call_index`` (the trace position where the operator asked). Highest index first so
     earlier inserts don't shift later ones. No structured options are attached (prose has none), which is safe:
     choice_respected only flags a provisioned partition that matches a NON-chosen OPTION label, so with no
-    options it cannot false-positive."""
-    todo = [e for e in (exchanges or []) if e.get("question")]
+    options it cannot false-positive.
+
+    A NUDGE (kind "nudge" — the ACP turn policy told a paused operator to carry on) is stamped as a ``user_nudge``
+    marker (like the runner's ``user_interjection``), NOT as an AskUserQuestion: the operator put nothing to the
+    user, so it must not count as a spend question answered — "Login node is up, next I'll provision" + "carry on"
+    would otherwise satisfy `spend_follows_question` through the broad spend-ish regex (`node`, `provision`)."""
+    todo = [e for e in (exchanges or []) if e.get("question") or e.get("kind") == "nudge"]
     for e in sorted(todo, key=lambda e: -int(e.get("call_index") or 0)):
         k = int(e.get("call_index") or 0)
         at = min(k + 1, len(trace.calls))
-        q = str(e["question"])
+        q = str(e.get("question") or "")
         a = str(e.get("answer") or "")
-        tc = ToolCall.of("AskUserQuestion", {"questions": [{"question": q}]}, answers={q: a})
+        if e.get("kind") == "nudge":
+            tc = ToolCall(name="user_nudge", input={"text": a}, result={"after_call": k, "paused_on": q[-300:]},
+                          raw_name="user_nudge")
+        else:
+            tc = ToolCall.of("AskUserQuestion", {"questions": [{"question": q}]}, answers={q: a})
         trace.calls.insert(at, tc)
     return trace