Standardized messages

AGENTS.md

@@ -17,14 +17,19 @@ If a tradeoff is required, choose correctness and robustness over short-term con
 - `apps/web`: React/Vite UI. Owns session UX, conversation/event rendering, and client-side state. Connects to the server via WebSocket.
 - `packages/contracts`: Shared Zod schemas and TypeScript contracts for provider events, WebSocket protocol, and model/session types.
 
+## Provider WS API Docs Sync (Required)
+- The canonical provider API spec lives at `docs/provider-ws-api.md`.
+- Any change to provider WS contracts, methods/channels, canonical event shapes, normalization mappings, replay/backpressure behavior, or client apply semantics must update `docs/provider-ws-api.md` in the same PR.
+- If code and docs diverge, treat it as a bug and resolve before merge.
+
 ## Codex App Server (Important)
 T3 Code is currently Codex-first. The server starts `codex app-server` (JSON-RPC over stdio) per provider session, then streams structured events to the browser through WebSocket push messages.
 
 How we use it in this codebase:
 - Session startup/resume and turn lifecycle are brokered in `apps/server/src/codexAppServerManager.ts`.
 - Provider dispatch and thread event logging are coordinated in `apps/server/src/providerManager.ts`.
 - WebSocket server routes NativeApi methods in `apps/server/src/wsServer.ts`.
-- Web app consumes provider event streams via WebSocket push on channel `providers.event`.
+- Web app consumes canonical provider stream frames via WebSocket push on channel `providers.stream`.
 
 Docs:
 - Codex App Server docs: https://developers.openai.com/codex/sdk/#app-server

README.md

@@ -76,18 +76,21 @@ T3 Code has a global runtime mode switch in the chat toolbar:
 
 ## Provider architecture
 
-The web app communicates with the server via WebSocket using a simple JSON-RPC-style protocol:
+The web app communicates with the server via WebSocket using a JSON-RPC-style request/response layer plus a single canonical provider stream:
 
 - **Request/Response**: `{ id, method, params }` → `{ id, result }` or `{ id, error }`
-- **Push events**: `{ type: "push", channel, data }` for streaming provider events
+- **Push stream**: `{ type: "push", channel: "providers.stream", data: ProviderStreamFrame }`
 
 Methods mirror the `NativeApi` interface defined in `@t3tools/contracts`:
 
 - `providers.startSession`, `providers.sendTurn`, `providers.interruptTurn`
-- `providers.respondToRequest`, `providers.stopSession`, `providers.listSessions`
+- `providers.respondToApproval`, `providers.stopSession`, `providers.listSessions`
+- `providers.openStream`, `providers.closeStream`
 - `shell.openInEditor`, `server.getConfig`
 
-Codex is the only implemented provider. `claudeCode` is reserved in contracts/UI.
+The stream contract is `snapshot + delta` with cursor replay (`afterSeq`) for reconnect safety. The server normalizes provider-native events (currently Codex raw app-server events) into canonical events so UI rendering stays provider-agnostic.
+
+See `docs/provider-ws-api.md` for the full canonical schema, event mapping, and replay/gap semantics.
 
 ## CI quality gates
 

apps/server/src/codexAppServerManager.ts

@@ -5,7 +5,7 @@ import readline from "node:readline";
 
 import type {
   ProviderApprovalDecision,
-  ProviderEvent,
+  ProviderRawEvent,
   ProviderRequestKind,
   ProviderSendTurnInput,
   ProviderSession,
@@ -135,7 +135,7 @@ export function isRecoverableThreadResumeError(error: unknown): boolean {
 }
 
 export interface CodexAppServerManagerEvents {
-  event: [event: ProviderEvent];
+  event: [event: ProviderRawEvent];
 }
 
 export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEvents> {
@@ -333,18 +333,18 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
     });
   }
 
-  async respondToRequest(
+  async respondToApproval(
     sessionId: string,
-    requestId: string,
+    approvalId: string,
     decision: ProviderApprovalDecision,
   ): Promise<void> {
     const context = this.requireSession(sessionId);
-    const pendingRequest = context.pendingApprovals.get(requestId);
+    const pendingRequest = context.pendingApprovals.get(approvalId);
     if (!pendingRequest) {
-      throw new Error(`Unknown pending approval request: ${requestId}`);
+      throw new Error(`Unknown pending approval request: ${approvalId}`);
     }
 
-    context.pendingApprovals.delete(requestId);
+    context.pendingApprovals.delete(approvalId);
     this.writeMessage(context, {
       id: pendingRequest.jsonRpcId,
       result: {
@@ -584,9 +584,12 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
   private handleServerRequest(context: CodexSessionContext, request: JsonRpcRequest): void {
     const route = this.readRouteFields(request.params);
     const requestKind = this.requestKindForMethod(request.method);
+    const isUserInputRequest = request.method === "item/tool/requestUserInput";
     let requestId: string | undefined;
-    if (requestKind) {
+    if (requestKind || isUserInputRequest) {
       requestId = randomUUID();
+    }
+    if (requestKind && requestId) {
       const pendingRequest: PendingApprovalRequest = {
         requestId,
         jsonRpcId: request.id,
@@ -621,11 +624,30 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
       return;
     }
 
-    if (request.method === "item/tool/requestUserInput") {
+    if (isUserInputRequest) {
       this.writeMessage(context, {
         id: request.id,
         result: { answers: {} },
       });
+
+      if (requestId) {
+        this.emitEvent({
+          id: randomUUID(),
+          kind: "notification",
+          provider: "codex",
+          sessionId: context.session.sessionId,
+          createdAt: new Date().toISOString(),
+          method: "item/requestApproval/decision",
+          threadId: route.threadId,
+          turnId: route.turnId,
+          itemId: route.itemId,
+          requestId,
+          payload: {
+            requestId,
+            decision: "accept",
+          },
+        });
+      }
       return;
     }
 
@@ -720,7 +742,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
     });
   }
 
-  private emitEvent(event: ProviderEvent): void {
+  private emitEvent(event: ProviderRawEvent): void {
     this.emit("event", event);
   }
 

apps/server/src/providerEventNormalizer.test.ts

@@ -0,0 +1,181 @@
+import type { CanonicalSessionState, ProviderRawEvent } from "@t3tools/contracts";
+import { describe, expect, it } from "vitest";
+
+import { ProviderEventNormalizer } from "./providerEventNormalizer";
+
+function makeRawEvent(overrides: Partial<ProviderRawEvent> = {}): ProviderRawEvent {
+  return {
+    id: "raw-1",
+    kind: "notification",
+    provider: "codex",
+    sessionId: "sess-1",
+    createdAt: "2026-02-10T00:00:00.000Z",
+    method: "turn/started",
+    payload: {
+      thread: { id: "thread-1" },
+      turn: { id: "turn-1" },
+    },
+    ...overrides,
+  };
+}
+
+function makeSession(overrides: Partial<CanonicalSessionState> = {}): CanonicalSessionState {
+  return {
+    sessionId: "sess-1",
+    provider: "codex",
+    status: "ready",
+    createdAt: "2026-02-10T00:00:00.000Z",
+    updatedAt: "2026-02-10T00:00:00.000Z",
+    ...overrides,
+  };
+}
+
+describe("ProviderEventNormalizer", () => {
+  it("maps turn start notifications to canonical session and turn events", () => {
+    const normalizer = new ProviderEventNormalizer();
+
+    const normalized = normalizer.normalize(
+      makeRawEvent({
+        method: "turn/started",
+      }),
+      makeSession(),
+    );
+
+    expect(normalized).toHaveLength(2);
+    expect(normalized[0]?.type).toBe("session.updated");
+    expect(normalized[1]?.type).toBe("turn.started");
+  });
+
+  it("maps command approval requests and decision notifications", () => {
+    const normalizer = new ProviderEventNormalizer();
+
+    const requested = normalizer.normalize(
+      makeRawEvent({
+        kind: "request",
+        method: "item/commandExecution/requestApproval",
+        requestId: "approval-1",
+        payload: {
+          thread: { id: "thread-1" },
+          turn: { id: "turn-1" },
+          command: "git status --short",
+        },
+      }),
+      makeSession(),
+    );
+
+    expect(requested.some((event) => event.type === "approval.requested")).toBe(true);
+
+    const resolved = normalizer.normalize(
+      makeRawEvent({
+        kind: "notification",
+        method: "item/requestApproval/decision",
+        requestId: "approval-1",
+        payload: {
+          requestId: "approval-1",
+          decision: "acceptForSession",
+        },
+      }),
+      makeSession(),
+    );
+
+    const approvalResolved = resolved.find((event) => event.type === "approval.resolved");
+    expect(approvalResolved).toBeDefined();
+    if (!approvalResolved || approvalResolved.type !== "approval.resolved") {
+      throw new Error("expected approval.resolved event");
+    }
+    expect(approvalResolved.decision).toBe("accept_for_session");
+  });
+
+  it("drops unmapped methods from the core stream", () => {
+    const normalizer = new ProviderEventNormalizer();
+
+    const normalized = normalizer.normalize(
+      makeRawEvent({
+        method: "item/reasoning/summaryPartAdded",
+        payload: {
+          text: "partial summary",
+        },
+      }),
+      makeSession(),
+    );
+
+    expect(normalized).toHaveLength(0);
+  });
+
+  it("maps message delta events when payload uses msg/message fallback fields", () => {
+    const normalizer = new ProviderEventNormalizer();
+
+    const normalized = normalizer.normalize(
+      makeRawEvent({
+        method: "item/agentMessage/delta",
+        threadId: undefined,
+        turnId: undefined,
+        itemId: undefined,
+        payload: {
+          msg: {
+            id: "msg-1",
+            thread_id: "thread-1",
+            turn_id: "turn-1",
+          },
+          delta: "hello",
+        },
+      }),
+      makeSession(),
+    );
+
+    const delta = normalized.find((event) => event.type === "message.delta");
+    expect(delta).toBeDefined();
+    if (!delta || delta.type !== "message.delta") {
+      throw new Error("expected message.delta event");
+    }
+
+    expect(delta.messageId).toBe("msg-1");
+    expect(delta.threadId).toBe("thread-1");
+    expect(delta.turnId).toBe("turn-1");
+    expect(delta.delta).toBe("hello");
+  });
+
+  it("maps assistant message completion with assistantMessage type aliases", () => {
+    const normalizer = new ProviderEventNormalizer();
+
+    const normalized = normalizer.normalize(
+      makeRawEvent({
+        method: "item/completed",
+        itemId: "msg_abc",
+        payload: {
+          threadId: "thread-1",
+          turnId: "turn-1",
+          item: {
+            id: "msg_abc",
+            type: "assistant_message",
+            text: "done",
+          },
+        },
+      }),
+      makeSession(),
+    );
+
+    const completed = normalized.find((event) => event.type === "message.completed");
+    expect(completed).toBeDefined();
+    if (!completed || completed.type !== "message.completed") {
+      throw new Error("expected message.completed event");
+    }
+
+    expect(completed.messageId).toBe("msg_abc");
+    expect(completed.text).toBe("done");
+  });
+
+  it("creates debug.raw wrappers", () => {
+    const normalizer = new ProviderEventNormalizer();
+    const raw = makeRawEvent({ method: "item/reasoning/summaryPartAdded" });
+
+    const debug = normalizer.toDebugRaw(raw);
+
+    expect(debug.type).toBe("debug.raw");
+    if (debug.type !== "debug.raw") {
+      throw new Error("expected debug.raw event");
+    }
+    expect(debug.provider).toBe("codex");
+    expect(debug.method).toBe("item/reasoning/summaryPartAdded");
+  });
+});