import { EventEmitter } from "node:events";import fs from "node:fs";import path from "node:path"; import {  type ProviderEvent,  type ProviderInterruptTurnInput,  type ProviderRespondToRequestInput,  type ProviderSendTurnInput,  type ProviderSession,  type ProviderSessionStartInput,  type ProviderStopSessionInput,  type ProviderTurnStartResult,  providerInterruptTurnInputSchema,  providerRespondToRequestInputSchema,  providerSendTurnInputSchema,  providerSessionStartInputSchema,  providerStopSessionInputSchema,} from "@t3tools/contracts";import { CodexAppServerManager } from "./codexAppServerManager"; export interface ProviderManagerEvents {  event: [event: ProviderEvent];} export class ProviderManager extends EventEmitter<ProviderManagerEvents> {  private readonly codex = new CodexAppServerManager();  private readonly threadLogsDir: string;  private readonly threadLogStreams = new Map<string, fs.WriteStream>();  private readonly sessionThreadIds = new Map<string, string>();  private readonly pendingEventsBySession = new Map<string, ProviderEvent[]>();  private disposed = false;  private readonly onCodexEvent = (event: ProviderEvent) => {    if (this.disposed) {      return;    }     this.routeEventToThreadLog(event);    this.emit("event", event);  };   constructor() {    super();     const logsDir = path.resolve(process.cwd(), ".logs");    this.threadLogsDir = path.join(logsDir, "threads");    fs.mkdirSync(this.threadLogsDir, { recursive: true });     this.codex.on("event", this.onCodexEvent);  }   async startSession(raw: ProviderSessionStartInput): Promise<ProviderSession> {    const input = providerSessionStartInputSchema.parse(raw);    if (input.provider !== "codex") {      throw new Error(`Provider '${input.provider}' is not implemented yet.`);    }     const session = await this.codex.startSession(input);    if (session.threadId) {      this.sessionThreadIds.set(session.sessionId, session.threadId);      this.flushPendingSessionEvents(session.sessionId, session.threadId);    }    return session;  }   async sendTurn(raw: ProviderSendTurnInput): Promise<ProviderTurnStartResult> {    const input = providerSendTurnInputSchema.parse(raw);    if (!this.codex.hasSession(input.sessionId)) {      throw new Error(`Unknown provider session: ${input.sessionId}`);    }     return this.codex.sendTurn(input);  }   async interruptTurn(raw: ProviderInterruptTurnInput): Promise<void> {    const input = providerInterruptTurnInputSchema.parse(raw);    if (!this.codex.hasSession(input.sessionId)) {      throw new Error(`Unknown provider session: ${input.sessionId}`);    }     await this.codex.interruptTurn(input.sessionId, input.turnId);  }   async respondToRequest(raw: ProviderRespondToRequestInput): Promise<void> {    const input = providerRespondToRequestInputSchema.parse(raw);    if (!this.codex.hasSession(input.sessionId)) {      throw new Error(`Unknown provider session: ${input.sessionId}`);    }     await this.codex.respondToRequest(input.sessionId, input.requestId, input.decision);  }   stopSession(raw: ProviderStopSessionInput): void {    const input = providerStopSessionInputSchema.parse(raw);    this.codex.stopSession(input.sessionId);    this.sessionThreadIds.delete(input.sessionId);    this.pendingEventsBySession.delete(input.sessionId);  }   listSessions(): ProviderSession[] {    return this.codex.listSessions();  }   stopAll(): void {    this.codex.stopAll();  }   dispose(): void {    if (this.disposed) {      return;    }     this.disposed = true;    this.codex.off("event", this.onCodexEvent);    for (const stream of this.threadLogStreams.values()) {      stream.end();    }    this.threadLogStreams.clear();    this.sessionThreadIds.clear();    this.pendingEventsBySession.clear();  }   private routeEventToThreadLog(event: ProviderEvent): void {    const threadId = this.resolveThreadId(event);    if (!threadId) {      const pending = this.pendingEventsBySession.get(event.sessionId) ?? [];      pending.push(event);      this.pendingEventsBySession.set(event.sessionId, pending.slice(-100));      return;    }     this.flushPendingSessionEvents(event.sessionId, threadId);    this.writeThreadEvent(threadId, event);  }   private flushPendingSessionEvents(sessionId: string, threadId: string): void {    const pending = this.pendingEventsBySession.get(sessionId);    if (!pending || pending.length === 0) {      return;    }     for (const event of pending) {      this.writeThreadEvent(threadId, event);    }    this.pendingEventsBySession.delete(sessionId);  }   private resolveThreadId(event: ProviderEvent): string | undefined {    const fromPayload = this.readThreadIdFromPayload(event.payload);    const threadId = event.threadId ?? fromPayload ?? this.sessionThreadIds.get(event.sessionId);     if (threadId) {      this.sessionThreadIds.set(event.sessionId, threadId);    }     return threadId;  }   private readThreadIdFromPayload(payload: unknown): string | undefined {    if (!payload || typeof payload !== "object") {      return undefined;    }     const record = payload as Record<string, unknown>;    const direct = record.threadId;    if (typeof direct === "string" && direct.length > 0) {      return direct;    }     const thread = record.thread;    if (!thread || typeof thread !== "object") {      return undefined;    }     const nested = (thread as Record<string, unknown>).id;    if (typeof nested === "string" && nested.length > 0) {      return nested;    }     return undefined;  }   private writeThreadEvent(threadId: string, event: ProviderEvent): void {    const stream = this.getOrCreateThreadLogStream(threadId);    stream.write(`${JSON.stringify(event)}\n`);  }   private getOrCreateThreadLogStream(threadId: string): fs.WriteStream {    const cached = this.threadLogStreams.get(threadId);    if (cached) {      return cached;    }     const safeThreadId = threadId.replace(/[^a-zA-Z0-9._-]/g, "_");    const stream = fs.createWriteStream(      path.join(this.threadLogsDir, `${safeThreadId}.events.ndjson`),      { flags: "a" },    );    this.threadLogStreams.set(threadId, stream);    return stream;  }}