1// Path joining only; the filesystem access lives in readContainedWorkflowFile.
2// @effect-diagnostics nodeBuiltinImport:off
3/**
4 * Reads what a workflow member actually answered.
5 *
6 * A member agent never becomes a provider task and emits nothing on the event
7 * stream, so the transcript the harness writes is the only full record of its
8 * turn — the progress snapshot carries a capped excerpt. The path is derived
9 * here from the run's transcript directory and the member's provider id rather
10 * than accepted from the caller, and the read itself is contained by
11 * readContainedWorkflowFile.
12 */
13import * as NodePath from "node:path";
14
15import { OrchestrationWorkflowFileError } from "@t3tools/contracts";
16import * as Effect from "effect/Effect";
17
18import { field, text } from "./unknownField.ts";
19import { readContainedWorkflowFile } from "./workflowFileRead.ts";
20
21const TRANSCRIPT_BYTE_CAP = 512 * 1024;
22
23/** A member id names a file, so anything path-shaped is rejected outright. */
24const AGENT_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/;
25
26/**
27 * Collects the assistant text of each turn in the harness's JSONL.
28 *
29 * Two shapes have to be handled: `attachment` lines and `isMeta` user lines are
30 * harness bookkeeping (hook output, skill listings, injected context) and are
31 * never conversation, and one assistant API message is split across several
32 * lines sharing a `message.id`, so those are joined into a single turn rather
33 * than read as separate ones.
34 */
35export function parseWorkflowAgentAnswers(contents: string): ReadonlyArray<string> {
36 const turns: string[] = [];
37 const turnIndexById = new Map<string, number>();
38
39 for (const line of contents.split("\n")) {
40 if (line.trim().length === 0) continue;
41 let entry: unknown;
42 try {
43 entry = JSON.parse(line);
44 } catch {
45 // A capped read can cut the final line mid-object; drop it and keep the rest.
46 continue;
47 }
48 if (text(field(entry, "type")) !== "assistant") continue;
49 if (field(entry, "isMeta") === true) continue;
50
51 const message = field(entry, "message");
52 const content = field(message, "content");
53 if (!Array.isArray(content)) continue;
54 const body = content
55 .filter((block) => text(field(block, "type")) === "text")
56 .map((block) => field(block, "text"))
57 .filter((value): value is string => typeof value === "string" && value.trim().length > 0)
58 .join("\n\n");
59 if (body.length === 0) continue;
60
61 const messageId = text(field(message, "id"));
62 const existing = messageId === undefined ? undefined : turnIndexById.get(messageId);
63 if (existing !== undefined) {
64 turns[existing] = `${turns[existing]}\n\n${body}`;
65 continue;
66 }
67 if (messageId !== undefined) turnIndexById.set(messageId, turns.length);
68 turns.push(body);
69 }
70
71 return turns;
72}
73
74export const readWorkflowAgentAnswers = Effect.fn("orchestration.readWorkflowAgentAnswers")(
75 function* (input: {
76 readonly transcriptDir: string;
77 readonly agentId: string;
78 readonly configDir?: string;
79 }) {
80 if (!AGENT_ID_PATTERN.test(input.agentId)) {
81 return yield* new OrchestrationWorkflowFileError({
82 reason: "invalid-path",
83 path: input.agentId,
84 });
85 }
86 const file = yield* readContainedWorkflowFile({
87 path: NodePath.join(input.transcriptDir, `agent-${input.agentId}.jsonl`),
88 extension: ".jsonl",
89 byteCap: TRANSCRIPT_BYTE_CAP,
90 tail: true,
91 ...(input.configDir === undefined ? {} : { configDir: input.configDir }),
92 });
93 return parseWorkflowAgentAnswers(
94 file.truncated ? file.contents.slice(file.contents.indexOf("\n") + 1) : file.contents,
95 );
96 },
97);