import { complete, getModel, type Api, type Model } from "@earendil-works/pi-ai";
import type { ModelRegistry } from "@earendil-works/pi-coding-agent";
import { buildPacket } from "./packet.ts";
import { OBSERVER_SYSTEM_PROMPT } from "./prompt.ts";
import { parseObserverOutput, OBSERVER_JSON_SCHEMA } from "./schema.ts";
import { liveThreads } from "./state.ts";
import type { AgentMessage } from "@earendil-works/pi-agent-core";
import type { ObserverOutput, Registry } from "./types.ts";
export interface ObserverConfig {
provider: string;
modelId: string;
apiKey?: string;
baseUrl?: string;
temperature?: number;
maxTokens?: number;
timeoutMs: number;
}
export type ObserverResult =
| { ok: true; output: ObserverOutput }
| { ok: false; error: string };
export function configFromEnv(env: NodeJS.ProcessEnv = process.env): ObserverConfig {
return {
provider: env.PRIME_COMMITMENT_OBSERVER_PROVIDER ?? "deepseek",
modelId: env.PRIME_COMMITMENT_OBSERVER_MODEL ?? "deepseek-v4-flash",
apiKey: env.PRIME_COMMITMENT_OBSERVER_API_KEY,
baseUrl: env.PRIME_COMMITMENT_OBSERVER_BASE_URL,
temperature: env.PRIME_COMMITMENT_OBSERVER_TEMPERATURE
? Number(env.PRIME_COMMITMENT_OBSERVER_TEMPERATURE)
: 0,
maxTokens: env.PRIME_COMMITMENT_OBSERVER_MAX_TOKENS
? Number(env.PRIME_COMMITMENT_OBSERVER_MAX_TOKENS)
: 4096,
timeoutMs: env.PRIME_COMMITMENT_OBSERVER_TIMEOUT_MS
? Number(env.PRIME_COMMITMENT_OBSERVER_TIMEOUT_MS)
: 60_000,
};
}
function resolveModel(registry: ModelRegistry, cfg: ObserverConfig): Model<Api> | undefined {
return registry.find(cfg.provider, cfg.modelId) ?? getModel(cfg.provider as never, cfg.modelId as never);
}
export async function runObserver(
registry: Registry,
messages: AgentMessage[],
targetMessageIds: ReadonlySet<string>,
modelRegistry: ModelRegistry,
cfg: ObserverConfig,
): Promise<ObserverResult> {
const packet = buildPacket(registry, messages, targetMessageIds);
if (packet.userMessages.length === 0 && packet.completedAssistantTurn.length === 0) {
return {
ok: true,
output: {
existing: liveThreads(registry).map(({ id, canonicalRequest, status }) => ({ id, canonicalRequest, status })),
additions: [],
},
};
}
const model = resolveModel(modelRegistry, cfg);
if (!model) return { ok: false, error: `Observer model ${cfg.provider}/${cfg.modelId} was not found` };
const auth = cfg.apiKey
? { ok: true as const, apiKey: cfg.apiKey, headers: undefined }
: await modelRegistry.getApiKeyAndHeaders(model);
if (!auth.ok) return { ok: false, error: auth.error };
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), cfg.timeoutMs);
try {
const result = await complete(
model,
{
systemPrompt: OBSERVER_SYSTEM_PROMPT,
messages: [
{
role: "user",
content: [{ type: "text", text: JSON.stringify(packet) }],
timestamp: Date.now(),
},
],
},
{
apiKey: auth.apiKey,
headers: auth.headers,
baseUrl: cfg.baseUrl,
temperature: cfg.temperature,
maxTokens: cfg.maxTokens,
signal: controller.signal,
responseFormat: {
type: "json_schema",
name: "commitment_observation",
strict: true,
schema: OBSERVER_JSON_SCHEMA,
} as never,
},
);
if (result.stopReason === "error" || result.stopReason === "aborted") {
return { ok: false, error: result.errorMessage ?? result.stopReason };
}
const text = result.content
.filter((part): part is Extract<(typeof result.content)[number], { type: "text" }> => part.type === "text")
.map((part) => part.text)
.join("");
const output = parseObserverOutput(
text,
new Map(packet.threads.map((thread) => [thread.id, thread.canonicalRequest])),
packet.userMessages.length > 0,
);
return output ? { ok: true, output } : { ok: false, error: "Observer returned invalid ledger JSON" };
} catch (error) {
return { ok: false, error: error instanceof Error ? error.message : String(error) };
} finally {
clearTimeout(timer);
}
}