diff --git a/crates/gjc-sdk/src/server.rs b/crates/gjc-sdk/src/server.rs index ec2788582d..8ea339b399 100644 --- a/crates/gjc-sdk/src/server.rs +++ b/crates/gjc-sdk/src/server.rs @@ -17,7 +17,7 @@ use std::{ path::PathBuf, sync::{ Arc, - atomic::{AtomicBool, AtomicU64, Ordering}, + atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, }, time::{Duration, Instant}, }; @@ -103,6 +103,12 @@ const CLIENT_HELLO_GRACE: Duration = Duration::from_secs(1); /// forced abort. const CONNECTION_JOIN_GRACE: Duration = Duration::from_secs(1); +/// Maximum host-directed frames waiting behind one connection writer. This +/// matches the positioned-event replay ring: once a subscriber falls farther +/// behind, rejecting new best-effort live sends keeps memory bounded and lets +/// replay report the authoritative sequence gap instead of buffering forever. +const MAX_QUEUED_DIRECTED_FRAMES: usize = 256; + /// Commands serialized through the owning connection task. #[derive(Debug)] enum DirectCommand { @@ -115,6 +121,19 @@ enum DirectCommand { ReevaluateAsk, } +fn reserve_directed_frame(counter: &AtomicUsize) -> bool { + counter + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |queued| { + (queued < MAX_QUEUED_DIRECTED_FRAMES).then_some(queued + 1) + }) + .is_ok() +} + +fn release_directed_frame(counter: &AtomicUsize) { + let queued = counter.fetch_sub(1, Ordering::Relaxed); + debug_assert!(queued > 0, "directed-frame reservation underflow"); +} + fn prepare_direct_ack(state: &ServerState, message: &ServerMessage) -> bool { let ServerMessage::AskSelectedAckRequest(request) = message else { return true; @@ -227,11 +246,12 @@ struct Delivered { #[derive(Debug, Clone)] struct Connection { - generation: String, - capabilities: Vec, - negotiation: Negotiation, - delivered: Option, - tx: mpsc::UnboundedSender, + generation: String, + capabilities: Vec, + negotiation: Negotiation, + delivered: Option, + queued_directed_frames: Arc, + tx: mpsc::UnboundedSender, } /// A rejected workflow-gate registration. @@ -735,7 +755,8 @@ impl ServerHandle { /// Send a validated JSON envelope to one connected v3 SDK client. Returns /// false when the destination is no longer current, the envelope is invalid, - /// or it exceeds the transport frame bound. + /// it exceeds the transport frame bound, or the connection's bounded writer + /// backlog is full. pub fn send_to(&self, connection_id: &str, json: String) -> bool { let Some((json, requires_tool_activity)) = validate_directed_frame(json) else { return false; @@ -745,15 +766,28 @@ impl ServerHandle { .connections .lock() .get(connection_id) - .map(|connection| (connection.tx.clone(), connection.generation.clone())); - sender.is_some_and(|(sender, connection_generation)| { - sender + .map(|connection| { + ( + connection.tx.clone(), + connection.generation.clone(), + Arc::clone(&connection.queued_directed_frames), + ) + }); + sender.is_some_and(|(sender, connection_generation, queued_directed_frames)| { + if !reserve_directed_frame(&queued_directed_frames) { + return false; + } + let sent = sender .send(DirectCommand::DirectedFrame { json, connection_generation, requires_tool_activity, }) - .is_ok() + .is_ok(); + if !sent { + release_directed_frame(&queued_directed_frames); + } + sent }) } @@ -1288,6 +1322,7 @@ async fn handle_conn(stream: TcpStream, state: Arc, cancel: Cancell format!("connection:{}", state.connection_sequence.fetch_add(1, Ordering::Relaxed)); let generation = "0".to_owned(); let (direct_tx, mut direct_rx) = mpsc::unbounded_channel::(); + let queued_directed_frames = Arc::new(AtomicUsize::new(0)); let mut rx = state.tx.subscribe(); let (mut write, mut read) = ws.split(); let hello = ServerMessage::Hello(ServerHello { @@ -1315,11 +1350,12 @@ async fn handle_conn(stream: TcpStream, state: Arc, cancel: Cancell .connections .lock() .insert(connection_id.clone(), Connection { - generation: generation.clone(), - capabilities: Vec::new(), - negotiation: Negotiation::AwaitingHello, - delivered: None, - tx: direct_tx.clone(), + generation: generation.clone(), + capabilities: Vec::new(), + negotiation: Negotiation::AwaitingHello, + delivered: None, + queued_directed_frames: Arc::clone(&queued_directed_frames), + tx: direct_tx.clone(), }); // Replay readiness before ask presentation; the ask itself is tailored by the @@ -1362,6 +1398,7 @@ async fn handle_conn(stream: TcpStream, state: Arc, cancel: Cancell connection_generation, requires_tool_activity, } => { + release_directed_frame(&queued_directed_frames); may_deliver_directed_frame( &state, &connection_id, @@ -1445,6 +1482,7 @@ async fn handle_conn(stream: TcpStream, state: Arc, cancel: Cancell connection_generation, requires_tool_activity, } => { + release_directed_frame(&queued_directed_frames); if may_deliver_directed_frame( &state, &connection_id, @@ -1876,6 +1914,60 @@ mod tests { // a paused runtime so concurrent libtest workers cannot share its clock. static PAUSED_TIME_TEST_LOCK: parking_lot::Mutex<()> = parking_lot::Mutex::new(()); + #[test] + fn directed_frame_reservations_are_bounded_and_reusable() { + let queued = AtomicUsize::new(0); + for _ in 0..MAX_QUEUED_DIRECTED_FRAMES { + assert!(reserve_directed_frame(&queued)); + } + assert!(!reserve_directed_frame(&queued)); + assert_eq!(queued.load(Ordering::Relaxed), MAX_QUEUED_DIRECTED_FRAMES); + + release_directed_frame(&queued); + assert!(reserve_directed_frame(&queued)); + assert_eq!(queued.load(Ordering::Relaxed), MAX_QUEUED_DIRECTED_FRAMES); + } + + #[tokio::test] + async fn directed_send_rejects_a_full_connection_writer_backlog() { + let handle = start(ServerConfig::new("s", "secret")).await.unwrap(); + let (tx, mut rx) = mpsc::unbounded_channel::(); + let queued = Arc::new(AtomicUsize::new(0)); + handle + .state + .connections + .lock() + .insert("slow".into(), Connection { + generation: "generation".into(), + capabilities: Vec::new(), + negotiation: Negotiation::Negotiated, + delivered: None, + queued_directed_frames: Arc::clone(&queued), + tx, + }); + + for id in 0..MAX_QUEUED_DIRECTED_FRAMES { + assert!( + handle + .send_to("slow", format!(r#"{{"type":"query_response","id":"q{id}","ok":true}}"#),) + ); + } + assert!( + !handle.send_to("slow", r#"{"type":"query_response","id":"overflow","ok":true}"#.into(),) + ); + assert_eq!(queued.load(Ordering::Relaxed), MAX_QUEUED_DIRECTED_FRAMES); + + let DirectCommand::DirectedFrame { .. } = rx.recv().await.expect("queued directed frame") + else { + panic!("expected directed frame"); + }; + release_directed_frame(&queued); + assert!( + handle.send_to("slow", r#"{"type":"query_response","id":"recovered","ok":true}"#.into(),) + ); + handle.stop(); + } + fn run_paused_test(test: impl std::future::Future) { let _time_guard = PAUSED_TIME_TEST_LOCK.lock(); tokio::runtime::Builder::new_current_thread() @@ -3617,6 +3709,7 @@ mod tests { capabilities: vec![capabilities::ASK_SELECTED_ACK_V1.into()], negotiation: Negotiation::Negotiated, delivered: None, + queued_directed_frames: Arc::new(AtomicUsize::new(0)), tx, }); let task = { diff --git a/crates/pi-natives/src/sdk.rs b/crates/pi-natives/src/sdk.rs index 92d6a58b7a..d0908a1ff1 100644 --- a/crates/pi-natives/src/sdk.rs +++ b/crates/pi-natives/src/sdk.rs @@ -702,7 +702,7 @@ impl NotificationServer { } else { Err(Error::from_reason( "SDK connection is unavailable or directed frame is invalid, oversized, or \ - unauthorized", + unauthorized, or its writer backlog is full", )) } } diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 7327cb7f32..3e13284167 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -3,6 +3,7 @@ ## [Unreleased] - Added the `commandcode-goat` model profile for the Command Code GOAT provider, assigning GLM-5.3 to the default role, DeepSeek V4 Flash to execution, Kimi K3 to planning, GLM-5.2 to criticism, and DeepSeek V4 Pro to architecture. +- Session endpoints hosted on the notification-adapter transport now deliver every ring-retained session event live to attached SDK subscribers as the same positioned `event` envelope (`generation`/`seq`) that `event_replay` returns, sent per connection over the validated directed leg with the same capability gating replay applies. Previously the live leg only pushed raw side-channel frames — the native broadcast enum reduced non-native kinds (including terminal `agent_end` lifecycle) to empty `unknown` frames, and correlated lifecycle reached only the submitting connection — so an already-attached direct SDK subscriber could observe a later positioned event, including a turn's terminal lifecycle, only by issuing another replay. Each connection's directed writer now bounds queued host frames to the replay-ring capacity; a lagged subscriber rejects additional best-effort live sends and recovers through replay (including the existing sequence-gap contract) instead of growing an unbounded backlog. Ring persistence, replay ordering, event positions, correlated requester delivery, and native notification frames are unchanged. - Telegram notification delivery now carries an explicit per-update inbound acknowledgement contract: user messages are acked `accepted` at session preflight acceptance (before the turn starts, so a fast turn can no longer out-race the pending-update registration), late admission failures ack `rejected`, and genuinely discarded frames ack `dropped`. Policy-suspended control commands are deferred to activation instead of being acked as dropped, per-update reaction transitions are serialized with terminal states monotonic (a slow queued 👀 can no longer overwrite a later ✅), and retraction sends the empty reaction list the Bot API requires. Daemon generation bumped 167→168. (#4528) - Managed fallback local snapshot failures now auto-recover with a bounded same-model retry (capped at `retry.maxRetries`) instead of terminating the turn: the discarded attempt is replay-safe and content-free, so the session re-issues the request without charging the provider fallback chain, advancing models, or mutating credentials. Exhausted retries still surface the explicit local diagnostic. diff --git a/packages/coding-agent/src/sdk/bus/index.ts b/packages/coding-agent/src/sdk/bus/index.ts index daacce0f90..95993be8a8 100644 --- a/packages/coding-agent/src/sdk/bus/index.ts +++ b/packages/coding-agent/src/sdk/bus/index.ts @@ -90,7 +90,14 @@ import { ensureBroker } from "../broker/ensure"; import { publishSessionHostRuntimeEvidence, type SessionHostRuntimePublication } from "../broker/lifecycle"; import { processIncarnation } from "../broker/process-incarnation"; import { SessionIndex } from "../broker/session-index"; -import { createSdkSurfaceFactory, type SessionSdkHost, SessionSdkSessionRuntime, shouldHostSdk } from "../host"; +import { + CAP_GATED_FRAME_KINDS, + createSdkSurfaceFactory, + type SessionSdkHost, + SessionSdkSessionRuntime, + shouldHostSdk, + TOOL_ACTIVITY_CAPABILITY, +} from "../host"; import { type AbortScope, type ControlSurface, dispatchControl, TypedControlError } from "../host/control"; import { BROKER_RUNTIME_CLOSE_CAPABILITY_FIELD } from "../host/control/runtime-gate"; import { CursorRegistry, QueryHandlers, RevisionStore, type SessionSurface } from "../host/query"; @@ -1134,6 +1141,9 @@ export class PresentationArbiter { interface SessionRuntime { server: NotificationServer; host: SessionSdkHost; + /** Delivers one ring-positioned event envelope to every attached subscriber + * connection, applying the same capability gate as event replay. */ + broadcastEventFrame: (event: SdkFrame) => void; /** Owns stateRoot-backed revisions and removes their spills on terminal shutdown. */ revisions: RevisionStore; /** Releases all snapshot pins before the revision store is closed. */ @@ -1349,11 +1359,22 @@ type SessionStartResult = { suppressExtensionError?: boolean; }; +/** Ring append plus positioned live broadcast. An event retained for replay + * must also reach already-attached subscribers live as the same positioned + * envelope, or they can only observe it by issuing another replay. */ +function emitSessionEvent( + runtime: Pick, + frame: { type: string; [key: string]: unknown }, + payload: Record = frame, +): void { + runtime.broadcastEventFrame(runtime.host.emitEvent({ kind: frame.type, payload })); +} + function pushSessionFrame( - runtime: Pick, + runtime: Pick, frame: { type: string; [key: string]: unknown }, ): void { - runtime.host.emitEvent({ kind: frame.type, payload: frame }); + emitSessionEvent(runtime, frame); if (frame.type === "turn_stream") { runtime.server.pushTurnStreamUnchecked( String(frame.sessionId), @@ -1368,30 +1389,30 @@ function pushSessionFrame( } async function pushTerminalSessionFrame( - runtime: Pick, + runtime: Pick, frame: { type: "session_closed"; sessionId: string }, ): Promise { - runtime.host.emitEvent({ kind: frame.type, payload: frame }); + emitSessionEvent(runtime, frame); return await runtime.server.pushFrameAndWait(JSON.stringify(frame), 1_000); } function pushFileAttachment( - runtime: Pick, + runtime: Pick, frame: { type: "file_attachment"; sessionId: string; name: string; mime?: string; caption?: string }, data: Buffer, ): void { - runtime.host.emitEvent({ kind: frame.type, payload: { ...frame, data: data.toString("base64") } }); + emitSessionEvent(runtime, frame, { ...frame, data: data.toString("base64") }); runtime.server.pushFileAttachmentUnchecked(frame.sessionId, frame.name, frame.mime, data, frame.caption); } /** Agent lifecycle is SDK session truth, independent of optional chat delivery. */ function emitAgentLifecycle( - runtime: Pick, + runtime: Pick, frame: { type: "agent_start" | "agent_end"; sessionId: string; commandId?: string; turnId?: string }, ): void { try { const json = JSON.stringify(frame); - runtime.host.emitEvent({ kind: frame.type, payload: frame }); + emitSessionEvent(runtime, frame); runtime.server.pushFrame(json); } catch (error) { logger.warn(`sdk: lifecycle delivery failed: ${String(error)}`); @@ -4625,6 +4646,29 @@ export function createNotificationsExtension( const promptSubmissions = new Map(); /** Connections fenced by a fatal prompt closure; their later frames are refused. */ const fencedConnections = new Set(); + /** + * Live positioned-event delivery to attached subscribers. The native + * broadcast channel round-trips a closed frame enum and cannot carry the + * positioned event envelope, so each envelope rides the validated directed + * leg instead — to every connection that completed capability negotiation + * or an event replay, which is exactly the attached-subscriber set. Fenced + * connections are excluded like their inbound frames, and capability-gated + * kinds follow the same gate replay applies, so live and replay delivery + * stay one truth per connection. + */ + const broadcastEventFrame = (event: SdkFrame): void => { + const gated = CAP_GATED_FRAME_KINDS.has(String(event.kind)); + const json = JSON.stringify(event); + for (const [connectionId, capabilities] of hostCapCache) { + if (fencedConnections.has(connectionId)) continue; + if (gated && !capabilities.has(TOOL_ACTIVITY_CAPABILITY)) continue; + try { + server.sendTo(connectionId, json); + } catch { + // Broadcasts are best effort; directed responses surface send failures. + } + } + }; let cancelPreflightsForConnection: ((connectionId: string) => Promise) | undefined; const promptTerminalTombstones = new Map(); // Authoritative bounded reconciliation state for canonical Q26 turn.result @@ -4733,7 +4777,7 @@ export function createNotificationsExtension( const key = promptSubmissionKey(correlation); const submission = promptSubmissions.get(key); if (!submission) return; - runtime.host.emitEvent({ kind: frame.type, payload: frame }); + emitSessionEvent(runtime, frame); if (submission.abandoned) { if (submission.terminal) finalizePrompt(key, correlation); return; @@ -6364,7 +6408,7 @@ export function createNotificationsExtension( }, start: async () => await server.start(), stop: async () => await server.stopAndWait(), - broadcastFrame: frame => server.pushFrame(JSON.stringify(frame)), + broadcastFrame: frame => broadcastEventFrame(frame), }, ...(preparesExistingThread ? { readiness: "deferred" as const } : {}), ...(activationGate ? { activationGate } : {}), @@ -6645,6 +6689,7 @@ export function createNotificationsExtension( runtime = { server, host, + broadcastEventFrame, revisions, cursors, id, @@ -7359,7 +7404,7 @@ export function createNotificationsExtension( sessionId: id, ...buildIdentity(ctx.cwd, ctx.sessionManager.getSessionName(), telegramTopicsEnabled()), }; - host.emitEvent({ kind: identityHeader.type, payload: identityHeader }); + emitSessionEvent(initializedRuntime, identityHeader); const endpoint = await sdkRuntime.startTransport(); initializedRuntime.notificationOwnerState = "ready"; if (notificationsEnabledForSession && settingsAvailable && settings) { @@ -7420,7 +7465,7 @@ export function createNotificationsExtension( sessionId: id, ...buildIdentity(ctx.cwd, sessionName, telegramTopicsEnabled()), }; - host.emitEvent({ kind: identity.type, payload: identity }); + emitSessionEvent(initializedRuntime, identity); server.pushFrame(JSON.stringify(identity)); }, 250); sessionNameObserver.unref?.(); @@ -7434,7 +7479,7 @@ export function createNotificationsExtension( sessionId: id, ...buildIdentity(ctx.cwd, sessionNameAfterStartup, telegramTopicsEnabled()), }; - host.emitEvent({ kind: identity.type, payload: identity }); + emitSessionEvent(initializedRuntime, identity); server.pushFrame(JSON.stringify(identity)); } const agentDir = lifecycleAgentDir ?? settings?.getAgentDir?.(); diff --git a/packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts b/packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts index c6c6dc09fe..2a10fd21ab 100644 --- a/packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts +++ b/packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts @@ -269,8 +269,12 @@ export const SDK_LIFECYCLE_ROUTER_PROTOCOL_VERSION = 1; * generator's post-fix manifest check always byte-compares the regenerated disk * manifest against the current tree, and auto-reaps pre-registry legacy stray * Telegram daemons (#4533). + * Generation 168 adds per-update inbound acknowledgement authority and monotonic + * reaction settlement for Telegram notification delivery (#4528). + * Generation 169 delivers every ring-positioned session event live through the + * bounded, capability-gated directed subscriber leg used by replay. */ -export const DAEMON_GENERATION = 168; +export const DAEMON_GENERATION = 169; /** * Serving-compatibility boundary for daemon lifecycle requests. Epoch 7 diff --git a/packages/coding-agent/src/sdk/host/host.ts b/packages/coding-agent/src/sdk/host/host.ts index 743c17cdb1..f8b8819986 100644 --- a/packages/coding-agent/src/sdk/host/host.ts +++ b/packages/coding-agent/src/sdk/host/host.ts @@ -85,8 +85,11 @@ export interface SessionSdkHostOptions extends HostEndpointAdapters { activationGate?: SessionActivationGate; } -const TOOL_ACTIVITY_CAPABILITY = "tool_activity_v2"; -const CAP_GATED_FRAME_KINDS = new Set(["tool_activity", "reasoning_summary"]); +/** Shared by the replay filter and transport live broadcasts: a connection + * must see the same capability-gated event kinds on both legs, or live and + * replay delivery diverge for the same subscriber. */ +export const TOOL_ACTIVITY_CAPABILITY = "tool_activity_v2"; +export const CAP_GATED_FRAME_KINDS: ReadonlySet = new Set(["tool_activity", "reasoning_summary"]); const EMPTY_CAPABILITIES: ReadonlySet = new Set(); /** Safe, identifier-free explanations for every refused activation status. */ diff --git a/packages/coding-agent/test/notifications-tool-activity.test.ts b/packages/coding-agent/test/notifications-tool-activity.test.ts index deacad064a..18a09ef3c3 100644 --- a/packages/coding-agent/test/notifications-tool-activity.test.ts +++ b/packages/coding-agent/test/notifications-tool-activity.test.ts @@ -38,6 +38,7 @@ interface SetupResult { ctx: never; frames: Frame[]; ws: WebSocket; + url: string; sessionId: string; token: string; } @@ -104,7 +105,7 @@ async function setup( ws.send(JSON.stringify({ type: "hello", protocolVersion: 3, capabilities: ["tool_activity_v2"] })); await sleep(50); await sleep(250); - return { handlers, ctx, frames, ws, sessionId, token, settings, controller }; + return { handlers, ctx, frames, ws, url, sessionId, token, settings, controller }; } async function setConfig( @@ -195,6 +196,126 @@ describe("SDK replay capability filter", () => { }); }); +test("live positioned tool events use the same negotiated capability gate as replay", async () => { + await withNotifications(async () => { + const capable = await setup(); + const legacyFrames: Frame[] = []; + const legacy = new WebSocket(`${capable.url}/?token=${encodeURIComponent(capable.token)}`); + sockets.push(legacy); + legacy.addEventListener("message", event => legacyFrames.push(JSON.parse(String((event as MessageEvent).data)))); + await new Promise((resolve, reject) => { + legacy.addEventListener("open", () => resolve()); + legacy.addEventListener("error", () => reject(new Error("websocket error"))); + }); + legacy.send(JSON.stringify({ type: "hello", protocolVersion: 3, capabilities: [] })); + await sleep(50); + + capable.ws.send(JSON.stringify({ type: "event_replay", id: "capable-attach", sinceSeq: 0 })); + legacy.send(JSON.stringify({ type: "event_replay", id: "legacy-attach", sinceSeq: 0 })); + await waitFor( + () => capable.frames.some(frame => frame.type === "event_replay_result" && frame.id === "capable-attach"), + "capable attachment replay", + ); + await waitFor( + () => legacyFrames.some(frame => frame.type === "event_replay_result" && frame.id === "legacy-attach"), + "legacy attachment replay", + ); + const capableReplay = capable.frames.find( + frame => frame.type === "event_replay_result" && frame.id === "capable-attach", + )!; + const legacyReplay = legacyFrames.find( + frame => frame.type === "event_replay_result" && frame.id === "legacy-attach", + )!; + + await capable.handlers.get("turn_start")!({ type: "turn_start" } as never, capable.ctx); + await capable.handlers.get("tool_execution_start")!( + { + type: "tool_execution_start", + toolCallId: "positioned-call", + toolName: "apply_patch", + args: {}, + } as never, + capable.ctx, + ); + await capable.handlers.get("tool_execution_end")!( + { + type: "tool_execution_end", + toolCallId: "positioned-call", + toolName: "apply_patch", + result: {}, + isError: false, + } as never, + capable.ctx, + ); + await waitFor( + () => + capable.frames.filter( + frame => + frame.type === "event" && + frame.kind === "tool_activity" && + typeof frame.seq === "number" && + frame.seq > Number(capableReplay.lastSeq), + ).length === 2, + "capability-gated live positioned tool events", + ); + await sleep(100); + + const capableLive = capable.frames.filter( + frame => + frame.type === "event" && + frame.kind === "tool_activity" && + typeof frame.seq === "number" && + frame.seq > Number(capableReplay.lastSeq), + ); + expect(capableLive.map(frame => (frame.payload as Record).phase)).toEqual([ + "started", + "completed", + ]); + expect( + legacyFrames.filter( + frame => + frame.type === "event" && + frame.kind === "tool_activity" && + typeof frame.seq === "number" && + frame.seq > Number(legacyReplay.lastSeq), + ), + ).toHaveLength(0); + + capable.ws.send( + JSON.stringify({ + type: "event_replay", + id: "capable-after", + sinceGeneration: capableReplay.generation, + sinceSeq: capableReplay.lastSeq, + }), + ); + legacy.send( + JSON.stringify({ + type: "event_replay", + id: "legacy-after", + sinceGeneration: legacyReplay.generation, + sinceSeq: legacyReplay.lastSeq, + }), + ); + await waitFor( + () => capable.frames.some(frame => frame.type === "event_replay_result" && frame.id === "capable-after"), + "capable parity replay", + ); + await waitFor( + () => legacyFrames.some(frame => frame.type === "event_replay_result" && frame.id === "legacy-after"), + "legacy parity replay", + ); + const capableAfter = capable.frames.find( + frame => frame.type === "event_replay_result" && frame.id === "capable-after", + )!; + const legacyAfter = legacyFrames.find( + frame => frame.type === "event_replay_result" && frame.id === "legacy-after", + )!; + expect((capableAfter.events as Frame[]).filter(frame => frame.kind === "tool_activity")).toEqual(capableLive); + expect((legacyAfter.events as Frame[]).filter(frame => frame.kind === "tool_activity")).toHaveLength(0); + }); +}, 30000); + async function withNotifications(run: () => Promise): Promise { const previous = process.env.GJC_NOTIFICATIONS; process.env.GJC_NOTIFICATIONS = "1"; diff --git a/packages/coding-agent/test/notifications-topic-registry.test.ts b/packages/coding-agent/test/notifications-topic-registry.test.ts index ace3e712e0..7838731692 100644 --- a/packages/coding-agent/test/notifications-topic-registry.test.ts +++ b/packages/coding-agent/test/notifications-topic-registry.test.ts @@ -756,7 +756,7 @@ test("preserves a no-provenance endpoint claim before a held create can stage it await creating; expect(reg.endpointAuthority(binding)).toEqual({ state: "unique", sessionId: "B" }); }); -test("publishes exact durable authority generation 168 at serving epoch 87", () => { +test("publishes exact durable authority generation 169 at serving epoch 87", () => { // Generation 58: parser-valid durable-fence promotion and rollback. // Generation 152: a thrown steady heartbeat renewal in the run loop is // contained instead of terminating the daemon (#4200). @@ -783,7 +783,11 @@ test("publishes exact durable authority generation 168 at serving epoch 87", () // the generator's post-fix manifest check byte-compares the regenerated disk // manifest against the current tree, and pre-registry legacy stray Telegram // daemons are auto-reaped (#4533). - expect(DAEMON_GENERATION).toBe(168); + // Generation 168: adds per-update inbound acknowledgement authority and + // monotonic reaction settlement for Telegram notification delivery (#4528). + // Generation 169: delivers every ring-positioned session event live through + // the bounded, capability-gated directed subscriber leg used by replay. + expect(DAEMON_GENERATION).toBe(169); expect(SERVING_EPOCH).toBe(87); }); test("archives pending topics into retained inactive records", async () => { diff --git a/packages/coding-agent/test/sdk-host-wiring.test.ts b/packages/coding-agent/test/sdk-host-wiring.test.ts index aa53c2b0f0..1cc3b8ba21 100644 --- a/packages/coding-agent/test/sdk-host-wiring.test.ts +++ b/packages/coding-agent/test/sdk-host-wiring.test.ts @@ -1280,30 +1280,67 @@ test("SDK host replays file attachment data as base64 while passing raw bytes to socket.addEventListener("open", () => resolve(), { once: true }); socket.addEventListener("error", () => reject(new Error("WS error")), { once: true }); }); + socket.send(JSON.stringify({ type: "event_replay", id: "file-attach", sinceGeneration: 1, sinceSeq: 0 })); + await waitFor( + () => frames.some(frame => frame.type === "event_replay_result" && frame.id === "file-attach"), + "file attachment subscriber replay", + ); + const attachmentReplay = frames.find( + frame => frame.type === "event_replay_result" && frame.id === "file-attach", + )!; + const attachmentCursor = Number(attachmentReplay.lastSeq); await expect(getTelegramFileSink(sessionId)!({ path: attachmentPath })).resolves.toEqual({ ok: true }); await waitFor(() => nativeData !== undefined, "raw N-API file attachment"); expect(nativeData).toBeInstanceOf(Buffer); expect(nativeData).toEqual(bytes); + await waitFor( + () => + frames.some( + frame => + frame.type === "event" && + (frame.payload as Record | undefined)?.type === "file_attachment" && + typeof frame.seq === "number" && + frame.seq > attachmentCursor, + ), + "live positioned file attachment", + ); + const liveAttachment = frames.find( + frame => + frame.type === "event" && + (frame.payload as Record | undefined)?.type === "file_attachment" && + typeof frame.seq === "number" && + frame.seq > attachmentCursor, + )!; - socket.send(JSON.stringify({ type: "event_replay", id: "file-replay", sinceGeneration: 1, sinceSeq: 0 })); + socket.send( + JSON.stringify({ + type: "event_replay", + id: "file-replay", + sinceGeneration: attachmentReplay.generation, + sinceSeq: attachmentCursor, + }), + ); await waitFor( () => frames.some(frame => frame.type === "event_replay_result" && frame.id === "file-replay"), "file replay", ); const replay = frames.find(frame => frame.type === "event_replay_result" && frame.id === "file-replay"); - expect(replay?.events).toEqual( - expect.arrayContaining([ - expect.objectContaining({ - payload: expect.objectContaining({ - type: "file_attachment", - sessionId, - name: "replay.bin", - data: bytes.toString("base64"), - }), - }), - ]), + const replayedAttachment = (replay?.events as Record[]).find( + frame => (frame.payload as Record | undefined)?.type === "file_attachment", ); + expect(replayedAttachment).toEqual(liveAttachment); + expect(liveAttachment).toMatchObject({ + type: "event", + kind: "file_attachment", + generation: attachmentReplay.generation, + payload: expect.objectContaining({ + type: "file_attachment", + sessionId, + name: "replay.bin", + data: bytes.toString("base64"), + }), + }); } finally { if (originalPushFileAttachmentUnchecked) { nativePrototype.pushFileAttachmentUnchecked = originalPushFileAttachmentUnchecked; @@ -1471,6 +1508,104 @@ test("SDK host replays event frames over direct v3 ingress and routes queries th }); }); +test("SDK host preserves positioned live order and replay parity for every attached direct subscriber", async () => { + const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "gjc-sdk-live-events-")); + dirs.push(cwd); + const sessionId = `sdk-${Date.now()}`; + process.env.GJC_NOTIFICATIONS = "1"; + const handlers = start(context(cwd, sessionId)); + const endpointFile = path.join(cwd, ".gjc", "state", "sdk", `${sessionId}.json`); + await waitFor(() => fs.existsSync(endpointFile), "SDK endpoint"); + const endpoint = JSON.parse(fs.readFileSync(endpointFile, "utf8")) as { url: string; token: string }; + const clients = await Promise.all([ + SdkClient.connect(endpoint.url, endpoint.token), + SdkClient.connect(endpoint.url, endpoint.token), + ]); + try { + const liveEvents: Record[][] = [[], []]; + for (const [index, client] of clients.entries()) { + client.onFrame(frame => { + if (frame.type === "event") liveEvents[index]!.push(frame); + }); + } + // Authoritative attachment: replay completes before the terminal event exists. + const replays = (await Promise.all( + clients.map(client => client.request({ type: "event_replay", sinceGeneration: 1, sinceSeq: 0 })), + )) as Array<{ ok: boolean; generation: number; lastSeq: number }>; + expect(replays.every(replay => replay.ok)).toBe(true); + expect(new Set(replays.map(replay => replay.generation))).toEqual(new Set([replays[0]!.generation])); + expect(new Set(replays.map(replay => replay.lastSeq))).toEqual(new Set([replays[0]!.lastSeq])); + const sessionContext = context(cwd, sessionId); + await handlers.get("agent_start")?.({ type: "agent_start" }, sessionContext); + await handlers.get("agent_end")?.({ type: "agent_end" }, sessionContext); + // Both already-attached subscribers must receive the later positioned terminal + // event live before any further query, replay, or reconnect is issued. + await waitFor( + () => + liveEvents.every((events, index) => + events.some(event => { + const payload = event.payload as Record | undefined; + return ( + payload?.type === "agent_end" && + typeof event.seq === "number" && + event.seq > replays[index]!.lastSeq + ); + }), + ), + "live positioned terminal events", + ); + + const liveLifecycle = liveEvents.map((events, index) => + events.filter(event => { + const payloadType = (event.payload as Record | undefined)?.type; + return ( + (payloadType === "agent_start" || payloadType === "agent_end") && + typeof event.seq === "number" && + event.seq > replays[index]!.lastSeq + ); + }), + ); + for (const [index, lifecycle] of liveLifecycle.entries()) { + expect(lifecycle.map(event => (event.payload as Record).type)).toEqual([ + "agent_start", + "agent_end", + ]); + const seqs = lifecycle.map(event => Number(event.seq)); + expect(seqs).toEqual([...seqs].sort((left, right) => left - right)); + expect(new Set(seqs).size).toBe(seqs.length); + expect(lifecycle).toEqual( + lifecycle.map(() => + expect.objectContaining({ + type: "event", + generation: replays[index]!.generation, + payload: expect.objectContaining({ sessionId }), + }), + ), + ); + } + expect(liveLifecycle[1]).toEqual(liveLifecycle[0]); + + const postLiveReplays = (await Promise.all( + clients.map((client, index) => + client.request({ + type: "event_replay", + sinceGeneration: replays[index]!.generation, + sinceSeq: replays[index]!.lastSeq, + }), + ), + )) as Array<{ events: Record[] }>; + for (const [index, replay] of postLiveReplays.entries()) { + const replayLifecycle = replay.events.filter(event => { + const payloadType = (event.payload as Record | undefined)?.type; + return payloadType === "agent_start" || payloadType === "agent_end"; + }); + expect(replayLifecycle).toEqual(liveLifecycle[index]); + } + } finally { + await Promise.all(clients.map(client => client.close())); + } +}); + test("SDK host preserves ordered prompt image blocks in the host payload", async () => { const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "gjc-sdk-prompt-images-")); dirs.push(cwd); diff --git a/packages/natives/CHANGELOG.md b/packages/natives/CHANGELOG.md index cfef17d639..429ba8fe8b 100644 --- a/packages/natives/CHANGELOG.md +++ b/packages/natives/CHANGELOG.md @@ -12,6 +12,7 @@ - `renameNoReplacePathAsync` and `linkNoReplacePathAsync`, async variants of the checked no-replace namespace publication primitives, scheduled on the native blocking-work pool so managed output publication can await the rename/link syscall boundary without blocking the host event loop (#4394). ### Fixed +- Bounded each SDK connection's queued host-directed frames to the positioned-event replay-ring capacity. Slow or stalled subscribers now reject excess directed sends for replay recovery instead of retaining an unbounded in-memory writer backlog; accepted frames remain serialized through the same connection writer. - `exactReplacePath` now retries transient Windows destination-sharing violations (#4330): when another handle denies delete sharing on the destination, the pre-mutation destination open is retried a bounded 30 × 100 ms before failing, and an exhausted retry reports the specific `sharing_violation` category with the underlying hex NTSTATUS (`windowsErrorCode`, e.g. `0xC0000043`) instead of a bare `io_error`. Only the destination open is retried — never after any namespace mutation — so a retry can never publish twice, and permission, path-not-found, disk-full, and identity failures are never retried. ## [0.13.2] - 2026-08-13 diff --git a/scripts/telegram-daemon-generation-manifest.json b/scripts/telegram-daemon-generation-manifest.json index c8b0052145..858a5fa0ca 100644 --- a/scripts/telegram-daemon-generation-manifest.json +++ b/scripts/telegram-daemon-generation-manifest.json @@ -526,7 +526,7 @@ "telegram:packages/coding-agent/src/sdk/bus/daemon-paths.ts:HEARTBEAT_TTL_MS": "62255b5467995d21d3f929c863278ca3e815102001c51ca6d66840e1522ff990", "telegram:packages/coding-agent/src/sdk/bus/daemon-paths.ts:daemonPaths": "1bd6ae51096fedb95d47149b977f8a40e7f814a774ecfce21e027aac268b0379", "telegram:packages/coding-agent/src/sdk/bus/index.ts:buildIdentity": "246ad10dd6341037a20379a8544736039584d36482f6b2d44e3054f5e7f87724", - "telegram:packages/coding-agent/src/sdk/bus/index.ts:createNotificationsExtension": "32faaf97a2b37aa7b273a6a6ed4b825c259e05a746bc8e391a77c03cd901d6b2", + "telegram:packages/coding-agent/src/sdk/bus/index.ts:createNotificationsExtension": "bd4ac284d6628297a17d6bf920ddb04a94b3e920562c61b334d4be933cec8ca9", "telegram:packages/coding-agent/src/sdk/bus/notification-service.ts:DaemonTransitionLock": "0fb018a6384bff312aab0345012936f7e0609e4691c841919426ad3d75841dcb", "telegram:packages/coding-agent/src/sdk/bus/notification-service.ts:NATIVE_PATH_IDENTITY_CONTRACT_VERSION": "ec669ef396909ce429e08ce4fa9a78b5f0106d8cedf967e634c6ae6974830b8a", "telegram:packages/coding-agent/src/sdk/bus/notification-service.ts:acquireDaemonTransitionLock": "0115500fcb5c5797008d607bd69694579c5b372420310f265d33a4759364decc", @@ -545,7 +545,7 @@ "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts:ownerPidFromOwnerId": "46691373b2bee01f28f3817a6aa6a7efffe880c2cea337c89155582c98d952bf", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts:runDaemonInternal": "3a65a0c0631214cc41679d379399c96c15bd9ed16802f772c031d35ae207d82a", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts:runDaemonSmoke": "6f085a667aa5c83de46d2d8945fb845c355fcbb43c46872342a44489203a5830", - "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts:DAEMON_GENERATION": "344735ee87440a8867fe69aa950a2a2a6931cf909c4bdebb5366a63907541643", + "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts:DAEMON_GENERATION": "e6b4c2d856582b6e34df2c2592a96e0ded4f808f6f131ec758d0dc4cc674f9bc", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts:NOTIFICATION_PROTOCOL_VERSION": "b99289f651fedcf020d28dbaf6f07dd37e7e4a5f6dc1f5118b872112325f1e81", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-control.ts:DaemonProcessReference": "c3d13e3670a6245a1250c4ebfcd80a36dd8fc96c67ab64d9f979182bd117bc4e", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-control.ts:TelegramDaemonController": "166b909a9073e4d1052b245152789cfb7a272cbf5fe3075e9e651766e68d0b1e",