feat(cognitive): emit TASK_PROGRESS at every step boundary — closes the streaming pipe

acc/agent.py

@@ -475,9 +475,55 @@ async def _handle_task(msg: object) -> None:
                 )
                 return
 
+            # Phase progress-emit — publish TASK_PROGRESS at every step
+            # boundary so the prompt pane (PR #19) can render live
+            # "agent thinking" lines.  Only build the callback when the
+            # inbound payload carries a task_id — otherwise there's
+            # nothing for downstream listeners to correlate against.
+            inbound_task_id = str(data.get("task_id", "") or "")
+            progress_callback = None
+            if inbound_task_id:
+                from acc.signals import (  # noqa: PLC0415
+                    SIG_TASK_PROGRESS,
+                    subject_task_progress,
+                )
+
+                def _publish_progress(ctx) -> None:
+                    """Sync callback fired by CognitiveCore /
+                    dispatch_invocations at each step boundary.  Schedules
+                    an async publish via ``create_task`` so the cognitive
+                    pipeline never blocks on NATS — fire-and-forget
+                    matches the operational tolerance for occasional lost
+                    progress events (operators care about forward motion,
+                    not ordering guarantees)."""
+                    payload = {
+                        "signal_type": SIG_TASK_PROGRESS,
+                        "task_id": inbound_task_id,
+                        "agent_id": self.agent_id,
+                        "collective_id": collective_id,
+                        "ts": time.time(),
+                        "progress": ctx.to_dict(),
+                    }
+                    try:
+                        asyncio.create_task(
+                            self.backends.signaling.publish(
+                                subject_task_progress(collective_id),
+                                payload,
+                            ),
+                            name=f"task-progress-{inbound_task_id[:8]}",
+                        )
+                    except Exception:
+                        logger.exception(
+                            "task_loop: failed to schedule TASK_PROGRESS "
+                            "publish (task_id=%s)", inbound_task_id,
+                        )
+
+                progress_callback = _publish_progress
+
             result = await self._cognitive_core.process_task(  # type: ignore[union-attr]
                 task_payload=data,
                 role=self._active_role,
+                progress_callback=progress_callback,
             )
 
             # Phase 4.4 — Capability dispatch.  Parse [SKILL:...] /
@@ -506,6 +552,10 @@ async def _handle_task(msg: object) -> None:
                         # bypass the queue entirely (cheap fast-path).
                         oversight_queue=self._oversight_queue,
                         task_id=str(data.get("task_id", "")),
+                        # Phase progress-emit — share the same callback
+                        # so the prompt pane sees a continuous progress
+                        # stream across the LLM steps + each invocation.
+                        progress_callback=progress_callback,
                     )
                     logger.info(
                         "task_loop: dispatched %d capability invocation(s) "

acc/capability_dispatch.py

@@ -205,6 +205,7 @@ async def dispatch_invocations(
     *,
     oversight_queue: "Any | None" = None,
     task_id: str = "",
+    progress_callback: "Any | None" = None,
 ) -> list[InvocationOutcome]:
     """Execute each parsed marker through the cognitive core.
 
@@ -233,13 +234,49 @@ async def dispatch_invocations(
             oversight item's ``task_id`` field so the TUI Compliance
             screen can correlate the gated invocation back to the
             originating TASK_ASSIGN.  Empty string is fine.
+        progress_callback: Optional sync callable that fires once per
+            invocation BEFORE dispatch.  Receives a
+            :class:`acc.progress.ProgressContext` whose
+            ``current_step`` is the 1-based index in *invocations*,
+            ``total_steps_estimated`` is the full list length, and
+            ``step_label`` reads ``"Calling skill:echo"`` /
+            ``"Calling mcp:fs.read"``.  The agent's task loop wraps
+            this to publish TASK_PROGRESS — operators see one
+            progress line per dispatched tool in the prompt-pane
+            transcript, in addition to the trace lines emitted from
+            the outcomes.  ``None`` (default) disables emission.
 
     Returns:
         One :class:`InvocationOutcome` per input marker, in the same
         order.  Empty input returns ``[]``.
     """
     outcomes: list[InvocationOutcome] = []
-    for inv in invocations:
+    total = len(invocations)
+    for idx, inv in enumerate(invocations, start=1):
+        if progress_callback is not None:
+            try:
+                from acc.progress import ProgressContext  # noqa: PLC0415
+                progress_callback(ProgressContext(
+                    current_step=idx,
+                    total_steps_estimated=total,
+                    step_label=f"Calling {inv.kind}:{inv.target}",
+                    elapsed_ms=0,
+                    estimated_remaining_ms=0,
+                    deadline_ms=0,
+                    confidence=0.5,
+                    confidence_trend="STABLE",
+                    llm_calls_so_far=0,
+                    tokens_in_so_far=0,
+                    tokens_out_so_far=0,
+                    token_budget_remaining=0,
+                    over_budget=False,
+                    over_token_budget=False,
+                ))
+            except Exception:
+                logger.exception(
+                    "capability_dispatch: progress callback raised "
+                    "for %s — continuing", inv.target,
+                )
         outcomes.append(await _dispatch_one(
             inv, core, role,
             oversight_queue=oversight_queue,

acc/cognitive_core.py

@@ -23,8 +23,16 @@
 
 from acc.config import ComplianceConfig, RoleDefinitionConfig
 from acc.governance_capabilities import CapabilityDecision, CapabilityGuard
+from acc.progress import ProgressContext
 from acc.signals import redis_centroid_key, redis_stress_key
 
+# Total steps in the canonical process_task pipeline (PRE-GATE → DRIFT).
+# Used as ``total_steps_estimated`` in every progress emission so the
+# operator's transcript shows a steady "step N/6" counter for the LLM
+# half of the work.  Capability dispatch (skills/MCPs) emits its own
+# progress with its own total — operators see two streams in sequence.
+_PROCESS_TASK_TOTAL_STEPS = 6
+
 logger = logging.getLogger("acc.cognitive_core")
 
 
@@ -284,6 +292,8 @@ async def process_task(
         self,
         task_payload: dict,
         role: Optional[RoleDefinitionConfig] = None,
+        *,
+        progress_callback: Optional[Any] = None,
     ) -> CognitiveResult:
         """Run the full reasoning pipeline for one task.
 
@@ -294,14 +304,63 @@ async def process_task(
             task_payload: TASK_ASSIGN signal payload dict.
             role: Active role definition. Falls back to empty RoleDefinitionConfig
                   if not provided.
+            progress_callback: Optional sync callable that fires once per
+                step boundary in the pipeline (PRE-GATE → DRIFT — six
+                emits in the happy path).  Receives a
+                :class:`acc.progress.ProgressContext`.  The agent's
+                task loop wraps this to publish TASK_PROGRESS on
+                ``acc.{cid}.task.progress`` so the prompt pane (PR #19)
+                renders live "agent thinking" lines.  ``None`` (default)
+                disables emission — zero overhead for non-prompt-pane
+                consumers.
 
         Returns:
             :class:`CognitiveResult` with updated :class:`StressIndicators`.
         """
         if role is None:
             role = RoleDefinitionConfig()
 
+        # Wall-clock anchor for ``elapsed_ms`` in every emission.
+        process_start_t = time.monotonic()
+
+        def _emit(step: int, label: str, *,
+                  confidence: float = 0.5,
+                  llm_calls: int = 0,
+                  tokens_in: int = 0,
+                  tokens_out: int = 0) -> None:
+            """Emit one progress step.  No-op when callback is None.
+
+            Exception-isolated so a misbehaving callback can't break
+            the cognitive pipeline — the operator-facing TUI surface
+            should never hold the agent's task loop hostage.
+            """
+            if progress_callback is None:
+                return
+            try:
+                ctx = ProgressContext(
+                    current_step=step,
+                    total_steps_estimated=_PROCESS_TASK_TOTAL_STEPS,
+                    step_label=label,
+                    elapsed_ms=int((time.monotonic() - process_start_t) * 1000),
+                    estimated_remaining_ms=0,
+                    deadline_ms=0,
+                    confidence=confidence,
+                    confidence_trend="STABLE",
+                    llm_calls_so_far=llm_calls,
+                    tokens_in_so_far=tokens_in,
+                    tokens_out_so_far=tokens_out,
+                    token_budget_remaining=0,
+                    over_budget=False,
+                    over_token_budget=False,
+                )
+                progress_callback(ctx)
+            except Exception:
+                logger.exception(
+                    "cognitive_core: progress callback raised at step=%d", step,
+                )
+
         # 1 — PRE-GATE
+        _emit(1, "Pre-reasoning gate (Cat-B setpoints)")
         blocked, block_reason = self._pre_reasoning_gate(role)
         if blocked:
             self._stress.cat_b_trigger_count += 1
@@ -319,6 +378,7 @@ async def process_task(
             )
 
         # 2 — PROMPT BUILD
+        _emit(2, "Building system prompt")
         system_prompt = self.build_system_prompt(role)
         user_content: str = task_payload.get("content", "")
 
@@ -375,7 +435,10 @@ async def process_task(
             except Exception as exc:
                 logger.warning("cognitive_core: Cat-A evaluation error: %s", exc)
 
-        # 3 — LLM CALL (async)
+        # 3 — LLM CALL (async).  Emit BEFORE the call — the LLM is the
+        # slowest part of the pipeline, so the operator sees the
+        # "Calling LLM" line stay visible for most of the elapsed time.
+        _emit(3, "Calling LLM")
         response, latency_ms, token_count = await self._call_llm(system_prompt, user_content)
         output_text: str = response.get("content", "")
 
@@ -386,11 +449,19 @@ async def process_task(
         else:
             self._stress.token_budget_utilization = 0.0
 
-        # 4 — POST-GATE
+        # 4 — POST-GATE.  Token counts now known — surface them in the
+        # progress event so the operator sees real numbers ticking up.
+        _emit(
+            4, "Post-reasoning governance",
+            llm_calls=1,
+            tokens_in=int(response.get("usage", {}).get("prompt_tokens", 0) or 0),
+            tokens_out=int(response.get("usage", {}).get("completion_tokens", 0) or 0),
+        )
         deviation_score = self._post_reasoning_governance(response, role)
         self._stress.cat_b_deviation_score += deviation_score
 
         # 5 — PERSIST episode (async embed)
+        _emit(5, "Persisting episode + embedding output")
         episode_id = ""
         output_embedding: list[float] = [0.0] * 384
         if output_text:
@@ -446,6 +517,7 @@ async def process_task(
         )
 
         # 6 — DRIFT (role centroid + domain centroid, ACC-11)
+        _emit(6, "Drift scoring")
         drift = await self._compute_drift(
             output_embedding, role,
             domain_centroid=self._domain_centroid or None,