1// @effect-diagnostics nodeBuiltinImport:off
2/** Workflow members appear only in transcripts, not provider events. */
3import * as NodePath from "node:path";
4
5import { OrchestrationWorkflowFileError } from "@t3tools/contracts";
6import * as Context from "effect/Context";
7import * as Effect from "effect/Effect";
8import * as Option from "effect/Option";
9import * as Schema from "effect/Schema";
10
11import { field, text } from "./unknownField.ts";
12import { readContainedWorkflowFile, type WorkflowFileIdentity } from "./workflowFileRead.ts";
13
14const TRANSCRIPT_READ_BYTE_CAP = 512 * 1024;
15
16const AGENT_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/;
17
18const decodeJsonLine = Schema.decodeUnknownOption(Schema.fromJsonString(Schema.Unknown));
19
20export interface WorkflowAgentTranscriptCursor {
21 readonly offset: number;
22 readonly identity: WorkflowFileIdentity | null;
23}
24
25export const initialWorkflowAgentTranscriptCursor: WorkflowAgentTranscriptCursor = {
26 offset: 0,
27 identity: null,
28};
29
30export const WorkflowAgentTranscriptReadObserver = Context.Reference<
31 (read: {
32 readonly agentId: string;
33 readonly startOffset: number;
34 readonly endOffset: number;
35 readonly lineCount: number;
36 }) => Effect.Effect<void>
37>("t3/orchestration/WorkflowAgentTranscriptReadObserver", {
38 defaultValue: () => () => Effect.void,
39});
40
41/** Skip harness bookkeeping (attachment and isMeta user lines) and lines without uuids needed for idempotent projection. */
42function parseWorkflowAgentTranscriptLines(contents: string): ReadonlyArray<unknown> {
43 return contents.split("\n").flatMap((line) => {
44 if (line.trim().length === 0) return [];
45 const entry = Option.getOrUndefined(decodeJsonLine(line));
46 const type = text(field(entry, "type"));
47 if (type !== "assistant" && type !== "user") return [];
48 if (field(entry, "isMeta") === true || text(field(entry, "uuid")) === undefined) return [];
49 return [entry];
50 });
51}
52
53/** Reads the complete lines written since `cursor`, and the cursor after them. */
54export const readWorkflowAgentTranscript = Effect.fn("orchestration.readWorkflowAgentTranscript")(
55 function* (input: {
56 readonly transcriptDir: string;
57 readonly agentId: string;
58 readonly cursor: WorkflowAgentTranscriptCursor;
59 readonly configDir?: string;
60 }) {
61 if (!AGENT_ID_PATTERN.test(input.agentId)) {
62 return yield* new OrchestrationWorkflowFileError({
63 reason: "invalid-path",
64 path: input.agentId,
65 });
66 }
67 const file = yield* readContainedWorkflowFile({
68 path: NodePath.join(input.transcriptDir, `agent-${input.agentId}.jsonl`),
69 extension: ".jsonl",
70 byteCap: TRANSCRIPT_READ_BYTE_CAP,
71 fromOffset: input.cursor,
72 ...(input.configDir === undefined ? {} : { configDir: input.configDir }),
73 });
74 const lines = parseWorkflowAgentTranscriptLines(file.contents);
75 yield* (yield* WorkflowAgentTranscriptReadObserver)({
76 agentId: input.agentId,
77 startOffset: file.startOffset,
78 endOffset: file.endOffset,
79 lineCount: lines.length,
80 });
81 return {
82 lines,
83 cursor: { offset: file.endOffset, identity: file.identity },
84 truncated: file.truncated,
85 };
86 },
87);