File size: 3,923 Bytes
bc575bc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | import { defineFacet, type Facet, type MutableReplicatedState } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import {
type AgentLane,
type HarnessEvent,
type LaneSnapshot,
type LaneTranscriptSnapshot,
type LaneWatchEvent,
reduceLaneSnapshot,
type WatchHandle,
} from "@earendil-works/pi-agent-core";
import { Transcript, type Transcript as TranscriptService, type TranscriptState } from "./transcript.ts";
interface TranscriptRuntime {
readonly service: TranscriptService;
activate(): Promise<void>;
dispose(): Promise<void>;
}
export function createTranscriptService(
lane: AgentLane,
createState: (initial: TranscriptState) => MutableReplicatedState<TranscriptState>,
): TranscriptRuntime {
const state = createState({ snapshot: null, event: null });
let watch: WatchHandle<LaneSnapshot> | undefined;
let rebase: Promise<void> | undefined;
let rebaseError: Error | undefined;
const publishSnapshot = (
next: LaneSnapshot,
event: LaneWatchEvent | null,
context: Parameters<typeof state.publish>[0],
): void => {
state.state.snapshot = next as LaneTranscriptSnapshot;
state.state.event = event;
state.publish(context);
};
const scheduleRebase = (context: Parameters<typeof state.publish>[0]): void => {
if (rebase !== undefined) return;
const activeWatch = watch;
if (activeWatch === undefined) return;
const pending = (async () => {
const refreshed = await activeWatch.resnapshot(context);
publishSnapshot(refreshed, null, context);
})();
rebase = pending;
void pending.then(
() => {
if (rebase === pending) rebase = undefined;
},
(error: unknown) => {
rebaseError = error instanceof Error ? error : new Error(String(error));
if (rebase === pending) rebase = undefined;
},
);
};
const onEvent = (event: HarnessEvent, context: Parameters<typeof state.publish>[0]): void => {
if (rebaseError !== undefined) throw rebaseError;
const forwarded = toLaneWatchEvent(event);
if (forwarded === undefined) return;
const snapshot = state.state.snapshot;
if (snapshot === null) throw new Error("Transcript service is not active");
if (reduceLaneSnapshot(snapshot, event) === "rebase") scheduleRebase(context);
state.state.event = forwarded;
state.publish(context);
};
return {
service: { state },
async activate() {
if (watch !== undefined) throw new Error("Transcript service is already active");
const opened = await lane.watch(BACKGROUND_CONTEXT);
watch = opened;
publishSnapshot(opened.snapshot, null, BACKGROUND_CONTEXT);
opened.start(onEvent);
},
async dispose() {
let failure: unknown;
try {
await rebase;
} catch (error) {
failure = error;
}
watch?.unsubscribe();
watch = undefined;
if (failure !== undefined) throw failure;
},
};
}
export function createTranscriptServiceFacet(lane: AgentLane): Facet {
return defineFacet({
id: "@pi/transcript",
setup(env) {
const runtime = createTranscriptService(lane, env.replicatedState);
env.provide(Transcript, runtime.service);
env.onActivate(() => runtime.activate());
env.own(() => runtime.dispose());
},
});
}
function toLaneWatchEvent(event: HarnessEvent): LaneWatchEvent | undefined {
switch (event.type) {
case "handler_error":
case "turn_start":
case "turn_end":
case "value_update":
case "lane_created":
return undefined;
case "config_update":
if (event.property !== "model" && event.property !== "thinkingLevel" && event.property !== "activeTools") {
return undefined;
}
return event as LaneWatchEvent;
case "message_update": {
if (event.message.role !== "assistant") {
throw new TypeError("Harness message_update did not carry an assistant message");
}
const { event: _providerEvent, ...update } = event;
return update as LaneWatchEvent;
}
default:
return event as LaneWatchEvent;
}
}
|