"""Turn a worker's progress into lines a person can read. Pure transformation: no files, no sockets, no clock. The runner decides where the lines go; this module decides what they say. Providers do not agree on a wire format, and this layer may not name one. So the formatter never reads a provider's JSON: it reads the normalised events below, and each provider adapter supplies the function that produces them from its own wire shape (``ExecCommand.normalise``). One wire shape is shared widely enough to be worth a default here — ``content_block_events``, whose events are keyed on ``type`` and carry ``message.content`` blocks — but it is a schema, not a provider, and an adapter that speaks something else says so in its own file. Screen output folds what the log keeps in full — a tool call becomes one line, a tool result becomes an outcome and a size. Thinking is dropped at normalisation, which is safe only because worker liveness is measured from stream arrival rather than from log writes (see the spec's 7.2). Three projections of the same stream: ``format_live`` for the pane, ``format_log`` for the archive, and ``final_text`` for the caller that must receive the closing message without the progress that produced it. """ from __future__ import annotations from dataclasses import dataclass from typing import Any, Callable, Mapping # Worker panes sit in a two-column grid whose narrowest allowed pane is the # floor named by the placement module (60 columns at the time of writing). # Holding a summary to two such lines keeps one tool call from pushing the # previous one off the top of a short pane. _MAX_SUMMARY = 120 @dataclass(frozen=True) class Text: """Prose the worker addressed to whoever is reading.""" body: str @dataclass(frozen=True) class ToolCall: """A tool the worker invoked, and the argument worth naming on one row.""" name: str detail: str = "" @dataclass(frozen=True) class ToolResult: """What a tool returned. ``size_bytes`` is carried rather than derived from ``body`` because the two answer different questions: the size is how much the tool produced, while the body is what the archive is willing to keep of it. ``failed`` is None when the stream closed the tool step without reporting an outcome at all. Reading that absence as success would put a confident "ok" next to a command that failed. """ body: str size_bytes: int failed: bool | None = None @dataclass(frozen=True) class Denial: """A tool call the provider refused before the worker could make it.""" tool: str reason: str @dataclass(frozen=True) class Result: """The worker's closing message. At most one per run.""" text: str StreamEvent = Text | ToolCall | ToolResult | Denial | Result # What an adapter hands the runner alongside its stream format: one line of the # CLI's output, already parsed, turned into however many normalised events it # carries. Returning nothing is how an event is dropped. Normalise = Callable[[Mapping[str, Any]], tuple[StreamEvent, ...]] def host_event_row( *, event_id: str, sequence: int, event_type: str, invocation_ref: str, attempt: int, payload: Mapping[str, Any], timestamp: str, ) -> dict[str, Any]: """Build one orchestrator-owned host-event row without I/O.""" row = { "schemaVersion": "1.0", "eventId": event_id, "sequence": sequence, "timestamp": timestamp, "eventType": event_type, "invocationRef": invocation_ref, "attempt": attempt, } row.update(dict(payload)) return row def no_events(event: Mapping[str, Any]) -> tuple[StreamEvent, ...]: """The normaliser for a CLI that emits no events at all. A text CLI's output is already what a person reads, so the runner forwards it whole and never calls this. It exists so ``ExecCommand`` can default to a truthful "there is nothing here to normalise" instead of to some provider's schema. """ return () def content_block_events(event: Mapping[str, Any]) -> tuple[StreamEvent, ...]: """Normalise the stream shape keyed on ``type`` with ``message.content``. Assistant events carry the worker's prose and its tool calls in the same content list, so one event becomes several normalised ones. Thinking blocks and everything unrecognised are dropped here rather than in the formatter: what is worth reading is a property of the stream, not of the projection. """ kind = event.get("type") if kind == "assistant": return tuple( item for block in _content_blocks(event) if (item := _assistant_event(block)) is not None ) if kind == "user": return tuple( _tool_result(block) for block in _content_blocks(event) if block.get("type") == "tool_result" ) if kind == "system" and event.get("subtype") == "permission_denied": return ( Denial( tool=str(event.get("tool_name", "?")), reason=str(event.get("decision_reason", "")).strip(), ), ) if kind == "result" and isinstance(event.get("result"), str): return (Result(text=str(event["result"])),) return () def format_live(event: StreamEvent) -> list[str]: return _rows(event, limit=_MAX_SUMMARY, include_body=False) def format_log(event: StreamEvent) -> list[str]: """What the archive keeps: everything the screen shows, plus the bodies. The log is no longer a live window — the worker pane is. It is read after the fact, by a person reconstructing why a worker failed and by ``okstra log-report``. """ return _rows(event, limit=None, include_body=True) def final_text(event: StreamEvent) -> str | None: """The worker's closing message, or None for any other event.""" return event.text if isinstance(event, Result) else None def _rows(event: StreamEvent, *, limit: int | None, include_body: bool) -> list[str]: if isinstance(event, Text): return _body_rows(event.body) if isinstance(event, ToolCall): head = f"→ {event.name}: {event.detail}" if event.detail else f"→ {event.name}" return [_truncate(head, limit)] if isinstance(event, ToolResult): rows = [f" ← {_outcome(event.failed)} ({event.size_bytes} bytes)"] if include_body: rows.extend(_body_rows(event.body)) return rows if isinstance(event, Denial): return [_truncate(f"!! PERMISSION DENIED — {event.tool}: {event.reason}", limit)] # A Result is printed by the runner at the point the run ends, not woven # into the progress it interrupts. return [] def _outcome(failed: bool | None) -> str: if failed is None: return "done" return "error" if failed else "ok" def _body_rows(text: str) -> list[str]: """One screen row per line, with every blank row dropped. Blank rows are dropped wherever they sit, not only at the end: a tool that double-spaces its output would otherwise take twice the pane height it earns, and a body that ends in newlines would pad the archive. """ return [part for part in text.splitlines() if part.strip()] def _assistant_event(block: Mapping[str, Any]) -> StreamEvent | None: if block.get("type") == "text": return Text(body=str(block.get("text", ""))) if block.get("type") == "tool_use": return ToolCall( name=str(block.get("name", "tool")), detail=_tool_detail(block.get("input")) ) return None def _tool_detail(payload: Any) -> str: if not isinstance(payload, Mapping): return "" for key in ("command", "file_path", "path", "pattern", "query"): value = payload.get(key) if value: return str(value) return "" def _tool_result(block: Mapping[str, Any]) -> ToolResult: body = block.get("content") return ToolResult( body=_body_text(body), size_bytes=_body_size(body), failed=bool(block.get("is_error")), ) def _body_text(body: Any) -> str: """The tool's own output as one string, whatever shape it arrived in.""" if isinstance(body, str): return body if isinstance(body, list): return "\n".join( str(block.get("text", "")) for block in body if isinstance(block, Mapping) ) return "" def _body_size(body: Any) -> int: """How much a tool returned, in bytes. The content arrives either as a plain string or as a list of content blocks. Counting only the string form reports a five-figure result as zero — a confident wrong number, which is worse on a screen than no number. Bytes rather than characters because the figure exists to be compared against the log this run leaves on disk. """ if isinstance(body, str): return len(body.encode("utf-8")) if isinstance(body, list): return sum( len(str(block.get("text", "")).encode("utf-8")) for block in body if isinstance(block, Mapping) ) return 0 def _content_blocks(event: Mapping[str, Any]) -> list[Mapping[str, Any]]: message = event.get("message") if not isinstance(message, Mapping): return [] content = message.get("content") if not isinstance(content, list): return [] return [block for block in content if isinstance(block, Mapping)] # A tool call's detail identifies itself at both ends and neither end alone. A # path's run-directory prefix is shared by every file a worker touches, so the # leaf is what tells two calls apart; a command's program name is at the front. # Cutting the tail served only the second, and on a project whose run directory # alone is 195 characters it rendered every file as the same visible string — # `→ Read: /Volumes/…/tasks/analysis-…` for all of them — leaving the reader # unable to tell one call from another. The head holds the tool name plus enough # of the detail to read a command; the rest of the budget goes to the tail. _HEAD_BUDGET = 32 def _truncate(line: str, limit: int | None) -> str: """Fold the middle, not the end. The end is what distinguishes one line from the next, so it is the part the screen must keep. Falls back to a tail cut only when the limit is too small to hold a head, an ellipsis, and any tail at all. """ if limit is None or len(line) <= limit: return line tail_budget = limit - _HEAD_BUDGET - 1 if tail_budget < 1: return line[: limit - 1] + "…" return line[:_HEAD_BUDGET] + "…" + line[-tail_budget:]