"""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, a size, and the first few rows of what came back. 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 # 화면이 받는 도구 결과 본문의 줄 수. 0 줄이던 동안 pane 에 남는 것은 성패와 # 바이트 수뿐이었다. 어떤 어댑터가 본문 대신 도구 이름표를 넘겨 모든 호출이 # 한 자릿수 바이트로 보이던 run 을, 화면만 보고는 아무도 판별하지 못했다. # 결과 하나가 pane 을 다 차지하면 직전 호출이 위로 밀려 나가므로 앞부분만 # 보여주고, 나머지는 로그가 갖는다. _LIVE_BODY_ROWS = 4 @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_describes_result`` marks a tool that reported *about* its result instead of returning it — a file read that answers ``54 lines, 4293 bytes`` rather than the 4293 bytes. Without the mark the two are the same shape and the size reads as the payload, so a 4293-byte read is archived as ``← done (20 bytes)`` and a reader concludes the read came back nearly empty. The provider that emits such a report says so; this layer only renders it differently. """ body: str size_bytes: int failed: bool | None = None body_describes_result: bool = False @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 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, body_rows=_LIVE_BODY_ROWS) 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, body_rows=None) 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, body_rows: int | None) -> 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): if event.body_describes_result: # 크기를 싣지 않는다. 이 이벤트가 가진 숫자는 보고 문자열의 길이일 # 뿐이고(보고조차 없으면 0), 그것을 payload 자리에 놓는 것이 오독의 # 출처였다. report = " ".join(event.body.split()) tail = ( f"the tool reported: {report}" if report else "the tool reported nothing" ) return [_truncate(f"← {_outcome(event.failed)} — no body returned; {tail}", limit)] head = f"← {_outcome(event.failed)} ({event.size_bytes} bytes)" return [head, *_result_body(event.body, limit=limit, keep=body_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 _result_body(body: str, *, limit: int | None, keep: int | None) -> list[str]: """결과 본문을 몇 줄 보여줄지 정한다. `keep is None` 이 기록의 몫이다 — 전부 남긴다. 화면은 앞 몇 줄만 받고, 생략한 줄 수를 밝힌다. 밝히지 않으면 짧은 출력과 잘린 출력이 같은 모양이 되어, 읽는 사람이 로그를 열 이유를 알 수 없다. """ rows = _body_rows(body) if keep is None: return rows shown = [_truncate(row, limit) for row in rows[:keep]] dropped = len(rows) - keep if dropped > 0: shown.append(f"… +{dropped} more line(s) — full body in the log") return shown 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:]