Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
### Fixed

- Slack and Discord chat daemons no longer post every lean final answer twice. The host sends each turn frame on a positioned leg and a raw leg and drops the raw copy only for connections that negotiate `positioned_notification_effects_v1`; the chat daemons' session connections never did, and a lean final carries no `messageRef` to collapse the two copies onto one publication. Chat daemon session connections now negotiate positioned-only notification effects, while plain SDK observers such as `gjc sdk session tail` keep both surfaces.
4 changes: 2 additions & 2 deletions packages/coding-agent/src/sdk/bus/chat-daemon-control.ts
Original file line number Diff line number Diff line change
Expand Up @@ -141,8 +141,8 @@ export type ChatDaemonAction = "stop" | "reload";
* crashes, recovery, and attachment retirement cannot replay ambiguous work.
*/
export const CHAT_DAEMON_GENERATIONS: Readonly<Record<ChatDaemonKind, number>> = {
discord: 82,
slack: 89,
discord: 83,
slack: 90,
};

export function chatDaemonGeneration(kind: ChatDaemonKind): number {
Expand Down
3 changes: 3 additions & 0 deletions packages/coding-agent/src/sdk/bus/chat-daemon-runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -318,6 +318,9 @@ export class ChatDaemonRuntime {
this.#router = new SessionRouter({
agentDir: input.agentDir,
observer: true,
// A lean final answer carries no messageRef, so the raw native copy of a
// positioned frame has no identity to collapse onto and would post twice.
positionedNotificationEffects: true,
correlateFrame,
deps: {
...deps.routerDeps,
Expand Down
2 changes: 1 addition & 1 deletion packages/coding-agent/src/sdk/bus/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,7 @@ import {
canDeliverSdkEvent,
createSdkSurfaceFactory,
masterAttestationForEffectiveHost,
POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY,
reattestMasterSessionIdentity,
SESSION_HOST_OBSERVER_CAPABILITY,
type SessionSdkHost,
Expand Down Expand Up @@ -204,7 +205,6 @@ import {
ASK_SELECTED_ACK_CAPABILITY,
type EnsureDaemonResult,
ensureTelegramDaemonRunningDetailed,
POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY,
} from "./telegram-daemon";

export type {
Expand Down
6 changes: 4 additions & 2 deletions packages/coding-agent/src/sdk/bus/telegram-daemon.ts
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ export {
TELEGRAM_TRANSPORT_GENERATION,
} from "./telegram-daemon-contract";

import { POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY } from "../host/host";
import {
type AgentDirSessionLifecycleService,
createSessionLifecycleService,
Expand Down Expand Up @@ -303,8 +304,9 @@ export const ASK_CONTROLS_CAPABILITY = "ask_controls_v1";
export const TOOL_ACTIVITY_CAPABILITY = "tool_activity_v2";
/** Receive-only compatibility capability for pre-v2 hosts. */
export const LEGACY_TOOL_ACTIVITY_CAPABILITY = "tool_activity_v1";
/** Opts notification adapters into positioned-only delivery for matching live effects. */
export const POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY = "positioned_notification_effects_v1";
/** Public `sdk/bus/telegram-daemon` path for the canonical `sdk/host` capability token. */
export { POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY };

type ToolActivityCapability = "v1" | "v2";

function negotiateToolActivityCapability(
Expand Down
6 changes: 6 additions & 0 deletions packages/coding-agent/src/sdk/host/host.ts
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,12 @@ export const TOOL_ACTIVITY_CAPABILITY = "tool_activity_v2";
export const TURN_STREAM_CAPABILITY = "turn_stream";
/** Capability used by notification adapters that observe a host without owning its work. */
export const SESSION_HOST_OBSERVER_CAPABILITY = "session_host_observer_v1";
/**
* Opts notification adapters into positioned-only delivery for matching live
* effects: the host excludes a negotiating connection from the raw native leg
* of every frame its positioned envelope already reached.
*/
export const POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY = "positioned_notification_effects_v1";
export const CAP_GATED_FRAME_KINDS: ReadonlySet<string> = new Set(["tool_activity", "reasoning_summary"]);
const EMPTY_CAPABILITIES: ReadonlySet<string> = new Set();

Expand Down
21 changes: 19 additions & 2 deletions packages/coding-agent/src/sdk/router/session-router.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,11 @@ import {
readSdkSessionEndpoint,
type SdkSessionEndpoint,
} from "../client/discovery";
import { SESSION_HOST_OBSERVER_CAPABILITY, TURN_STREAM_CAPABILITY } from "../host/host";
import {
POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY,
SESSION_HOST_OBSERVER_CAPABILITY,
TURN_STREAM_CAPABILITY,
} from "../host/host";
import {
type ActivatedPreparedSession,
type PreparedSessionActivationClient,
Expand Down Expand Up @@ -291,6 +295,13 @@ export interface SessionRouterOptions {
attachFilter?: (sessionId: string) => boolean;
/** Marks default SDK clients as notification-only observers; demanding by default. */
observer?: boolean;
/**
* Negotiates positioned-only notification effects on default SDK clients, so the
* host stops sending them the raw native copy of frames whose positioned
* envelope already reached them. Only notification-effect publishers opt in;
* ordinary SDK observers keep both public surfaces.
*/
positionedNotificationEffects?: boolean;
deps?: SessionRouterDeps;
/** Runtime-specific identity validation; Router supplies a conservative fallback. */
correlateFrame?: SessionRouterFrameCorrelator;
Expand Down Expand Up @@ -619,6 +630,7 @@ export class SessionRouter {
readonly #sessionIds: ReadonlySet<string> | undefined;
readonly #attachFilter: ((sessionId: string) => boolean) | undefined;
readonly #observer: boolean;
readonly #positionedNotificationEffects: boolean;
readonly #deps: SessionRouterDeps;
readonly #correlateFrame: SessionRouterFrameCorrelator;
readonly #index: SessionIndex;
Expand Down Expand Up @@ -662,6 +674,7 @@ export class SessionRouter {
this.#sessionIds = options.sessionIds === undefined ? undefined : new Set(options.sessionIds);
this.#attachFilter = options.attachFilter;
this.#observer = options.observer === true;
this.#positionedNotificationEffects = options.positionedNotificationEffects === true;
this.#deps = options.deps ?? {};
this.#correlateFrame = options.correlateFrame ?? fallbackCorrelation;
this.#index = this.#deps.createIndex?.(options.agentDir) ?? new DefaultSessionIndex(options.agentDir);
Expand Down Expand Up @@ -1530,9 +1543,13 @@ export class SessionRouter {
endpointMtimeMs,
});
} else {
const capabilities = [
...(this.#observer ? [SESSION_HOST_OBSERVER_CAPABILITY, TURN_STREAM_CAPABILITY] : []),
...(this.#positionedNotificationEffects ? [POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY] : []),
];
const defaultClient = new SdkClient(endpoint.url, endpoint.token, {
...ACP_SESSION_RECONNECT,
...(this.#observer ? { capabilities: [SESSION_HOST_OBSERVER_CAPABILITY, TURN_STREAM_CAPABILITY] } : {}),
...(capabilities.length > 0 ? { capabilities } : {}),
});
transport = defaultClient;
connection = defaultClient.connect().then(() => defaultClient);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,8 @@ import * as fs from "node:fs";
import * as os from "node:os";
import * as path from "node:path";
import { createNotificationsExtension } from "../src/sdk/bus/index";
import { POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY, TOOL_ACTIVITY_CAPABILITY } from "../src/sdk/bus/telegram-daemon";
import { TOOL_ACTIVITY_CAPABILITY } from "../src/sdk/bus/telegram-daemon";
import { POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY } from "../src/sdk/host/host";
import {
cleanupFixtureRoots,
createNotificationFixtureRoot,
Expand Down
110 changes: 110 additions & 0 deletions packages/coding-agent/test/sdk-chat-daemon-worker.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -265,6 +265,96 @@ class FakeSdkClient implements SessionRouterClient {
}
}

/**
* Drives one lean final answer through the production host into a real chat
* runtime and returns how many provider posts carry it. The host emits every
* turn_stream on a positioned leg and a raw native leg, excluding from the raw
* leg only connections that negotiated positioned-only notification effects. A
* lean final carries no messageRef, so a daemon that receives both legs has no
* shared identity to collapse them into one post.
*/
async function deliverLeanFinalThroughProductionHost(
root: string,
kind: "slack" | "discord",
finalText: string,
): Promise<number> {
const agentDir = path.join(root, ".gjc", "agent");
const host = await startProductionSdkHost(root);
const index = await new SessionIndex(agentDir).open();
const slackProvider = new FakeSlackProvider();
const discordProvider = new FakeDiscordProvider();
const notifications =
kind === "slack"
? {
slack: {
botToken: "bot-token",
appToken: "app-token",
workspaceId: "team",
channelId: "channel",
authorizedUserId: "human",
},
}
: {
discord: {
botToken: "discord-token",
applicationId: discordProvider.applicationId,
guildId: "guild",
parentChannelId: "parent",
},
};
const runtime = new ChatDaemonRuntime(
{
kind,
agentDir,
config: { identity: "fingerprint-only", presentation: { redact: false, verbosity: "lean" }, notifications },
},
{
createSlackProvider: () => slackProvider,
createDiscordProvider: () => discordProvider,
routerDeps: {
createIndex: () => index,
setInterval: (() => 0) as unknown as typeof setInterval,
clearInterval: (() => {}) as typeof clearInterval,
},
},
);
const count = (): number =>
kind === "slack"
? slackProvider.posts.filter(post => post.text.includes(finalText)).length
: discordProvider.messages.filter(message => message.content.includes(finalText)).length;
try {
await index.append({
type: "host_registered",
sessionId: host.sessionId,
locator: { cwd: root, worktreeRoot: null, stateRoot: path.join(root, ".gjc", "state") },
endpointGeneration: 1,
pid: host.endpoint.pid,
endpointMtimeMs: host.endpointMtimeMs,
});
await withStageTimeout(`${kind} runtime start`, runtime.start());
await withStageTimeout(`${kind} runtime admission`, runtime.reconcile({ waitForReplay: true }));
const runner = host.session.extensionRunner;
const assistant = { role: "assistant", content: finalText };
await runner?.emit({ type: "agent_start" });
await runner?.emit({ type: "message_end", message: assistant } as never);
await runner?.emit({ type: "turn_end", turnIndex: 0, message: assistant } as never);
await runner?.emit({
type: "agent_end",
stopReason: "completed",
messages: [{ role: "assistant", stopReason: "stop" }],
} as never);
const deadline = Date.now() + 10_000;
while (count() < 1 && Date.now() < deadline) await Bun.sleep(10);
// Both host legs land within milliseconds; leave room for a second
// representation to arrive before asserting that it never does.
await Bun.sleep(500);
await withStageTimeout(`${kind} runtime stop`, runtime.stop());
} finally {
await withStageTimeout("production SDK host stop", host.stop());
}
return count();
}

describe("chat daemon worker", () => {
let root = "";
afterEach(async () => {
Expand Down Expand Up @@ -1709,4 +1799,24 @@ describe("chat daemon worker", () => {
},
60_000,
);

it.skipIf(skipProductionSessionHost)(
"posts a lean final answer from the production host to Slack exactly once",
async () => {
root = await fs.mkdtemp(path.join(process.env.TMPDIR ?? "/tmp", "gjc-slack-lean-final-"));
const posts = await deliverLeanFinalThroughProductionHost(root, "slack", "Lean Slack final answer.");
expect(posts).toBe(1);
},
60_000,
);

it.skipIf(skipProductionSessionHost)(
"posts a lean final answer from the production host to Discord exactly once",
async () => {
root = await fs.mkdtemp(path.join(process.env.TMPDIR ?? "/tmp", "gjc-discord-lean-final-"));
const posts = await deliverLeanFinalThroughProductionHost(root, "discord", "Lean Discord final answer.");
expect(posts).toBe(1);
},
60_000,
);
});
53 changes: 52 additions & 1 deletion packages/coding-agent/test/sdk-host-wiring.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,11 @@ import { getTelegramFileSink } from "../src/sdk/bus/attachment-registry";
import { reconciliationStorePath } from "../src/sdk/bus/reconciliation-store";
import type { NotificationSessionController } from "../src/sdk/bus/session-control";
import { SdkClient } from "../src/sdk/client";
import { SESSION_HOST_OBSERVER_CAPABILITY, SessionSdkHost } from "../src/sdk/host";
import {
POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY,
SESSION_HOST_OBSERVER_CAPABILITY,
SessionSdkHost,
} from "../src/sdk/host";
import { createSdkRunCapability } from "../src/sdk/host/sdk-run-capability";
import { type SessionAttachment, SessionRouter } from "../src/sdk/router/session-router";
import { createAgentSession } from "../src/sdk/session";
Expand Down Expand Up @@ -1282,6 +1286,53 @@ test("a default ACP-shaped SessionRouter client remains demanding", async () =>
}
}, 60_000);

test("only a positioned-effects SessionRouter negotiates positioned-only notification delivery", async () => {
const negotiations: string[][] = [];
const negotiatedImpl = NotificationServer.prototype.onNegotiatedCapabilities;
const negotiatedHook = spyOn(NotificationServer.prototype, "onNegotiatedCapabilities").mockImplementation(function (
this: NotificationServer,
callback,
) {
return negotiatedImpl.call(this, (error, connectionId, capabilities) => {
// The threadsafe callback delivers its tuple as the second argument at runtime.
const negotiated = Array.isArray(connectionId) ? (connectionId as unknown[])[1] : capabilities;
if (Array.isArray(negotiated)) negotiations.push(negotiated.map(String));
return callback(error, connectionId, capabilities);
});
});
const cwd = fs.mkdtempSync(path.join(os.tmpdir(), "gjc-sdk-router-positioned-effects-"));
dirs.push(cwd);
const agentDir = path.join(cwd, ".gjc", "agent");
let host: Awaited<ReturnType<typeof startProductionSdkHost>> | undefined;
try {
host = await startProductionSdkHost(cwd, { acceptPromptPreflightWithoutExecution: true });
// A plain observer (e.g. `sdk session tail`) keeps both public surfaces (#4570).
const observer = new SessionRouter({ agentDir, observer: true });
try {
await observer.start();
await waitFor(() => negotiations.length === 1, "plain observer hello negotiation");
} finally {
await observer.stop();
}
expect(negotiations[0]).toContain(SESSION_HOST_OBSERVER_CAPABILITY);
expect(negotiations[0]).not.toContain(POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY);

const publisher = new SessionRouter({ agentDir, observer: true, positionedNotificationEffects: true });
try {
await publisher.start();
await waitFor(() => negotiations.length === 2, "positioned-effects hello negotiation");
} finally {
await publisher.stop();
}
expect(negotiations[1]).toEqual(
expect.arrayContaining([SESSION_HOST_OBSERVER_CAPABILITY, POSITIONED_NOTIFICATION_EFFECTS_CAPABILITY]),
);
} finally {
await host?.stop();
negotiatedHook.mockRestore();
}
}, 60_000);

test("late capability and replay callbacks after close cannot erase a live demanding client", async () => {
let negotiated: Parameters<NotificationServer["onNegotiatedCapabilities"]>[0] | undefined;
let closed: Parameters<NotificationServer["onConnectionClose"]>[0] | undefined;
Expand Down
8 changes: 4 additions & 4 deletions scripts/telegram-daemon-generation-manifest.json
Original file line number Diff line number Diff line change
Expand Up @@ -355,7 +355,7 @@
"discord:packages/coding-agent/src/sdk/bus/chat-daemon-cli.ts:loadConfig": "aff409ff6d08bfe19f5c39aac900ec6c8e898d988971bac029edb50459a84f5e",
"discord:packages/coding-agent/src/sdk/bus/chat-daemon-cli.ts:ownerPid": "31110dcdd6e0f5dbc8b4dce27383646b9c339739758623ab67dc40ddc36661e0",
"discord:packages/coding-agent/src/sdk/bus/chat-daemon-cli.ts:runChatDaemonInternal": "75b90374711fb7060161007debc1907e1e9a919ac9e89c46ab0c55542ee2f43d",
"discord:packages/coding-agent/src/sdk/bus/chat-daemon-control.ts:CHAT_DAEMON_GENERATIONS.discord": "a8859a9326999dafd1c5e3ca2304992432b79152f58c9cdf2922a78217e67e66",
"discord:packages/coding-agent/src/sdk/bus/chat-daemon-control.ts:CHAT_DAEMON_GENERATIONS.discord": "5d09346b12df473a4abc8dc4d9e8d28bc5b3704ffc429be131b8fa7856715094",
"discord:packages/coding-agent/src/sdk/bus/chat-daemon-control.ts:ChatDaemonAction": "d8acaf90439e410595b5cd56fd187001393ea943517a5563a4a2ea3c88d155ef",
"discord:packages/coding-agent/src/sdk/bus/chat-daemon-control.ts:ChatDaemonController": "bb722a6a83eb383fce45e756cba356592b33ff0f505b4263409ce9138ea651f1",
"discord:packages/coding-agent/src/sdk/bus/chat-daemon-control.ts:ChatDaemonKind": "b1c2906c4eb04e120c9ce42f8b549a68cfc834d29d4d745335712a7f61bf09f2",
Expand Down Expand Up @@ -433,13 +433,13 @@
"discord:packages/coding-agent/src/sdk/bus/discord-daemon.ts:stop": "c8bcd881cc0331fb0565a82d997805f939701b12a80e03177192dbd5935824ff",
"discord:packages/coding-agent/src/sdk/client/discovery.ts:readSdkSessionEndpoint": "bd8efa5f581993b3dc6c5549f62732211181b460b6e38d97220bb77b1875075e",
"discord:packages/coding-agent/src/sdk/router/session-router.ts:SessionRouter.#attach": "a08b206ed46659b003ff331708cfccdf81c5c3037263bacca71d675fd09321e3",
"discord:packages/coding-agent/src/sdk/router/session-router.ts:SessionRouter.#createAttachedClient": "76a3018aaf6f9b1dec802a884cb1e40c16a9c9b5ae5b842a70c1fc01c3cf53eb",
"discord:packages/coding-agent/src/sdk/router/session-router.ts:SessionRouter.#createAttachedClient": "f7d9879aeb8e84d4b4090567694b8332fdea95b25a79614e639a313dfd00fc87",
"discord:packages/coding-agent/src/sdk/router/session-router.ts:SessionRouter.#publishAttachment": "d46fff88a22ebd42e3e439a5cc20059a2ed002945972849e841a2e7d644a3079",
"slack:packages/coding-agent/src/sdk/bus/chat-daemon-cli.ts:defaultPidAlive": "437658a2f14ac5daa2a3e84dcd637078566e71a88841cd750efab6d48fc94c3f",
"slack:packages/coding-agent/src/sdk/bus/chat-daemon-cli.ts:loadConfig": "aff409ff6d08bfe19f5c39aac900ec6c8e898d988971bac029edb50459a84f5e",
"slack:packages/coding-agent/src/sdk/bus/chat-daemon-cli.ts:ownerPid": "31110dcdd6e0f5dbc8b4dce27383646b9c339739758623ab67dc40ddc36661e0",
"slack:packages/coding-agent/src/sdk/bus/chat-daemon-cli.ts:runChatDaemonInternal": "75b90374711fb7060161007debc1907e1e9a919ac9e89c46ab0c55542ee2f43d",
"slack:packages/coding-agent/src/sdk/bus/chat-daemon-control.ts:CHAT_DAEMON_GENERATIONS.slack": "dcf417b4588584fa316a85fdae4ca5df3775fba4fde23ad36764a66577193536",
"slack:packages/coding-agent/src/sdk/bus/chat-daemon-control.ts:CHAT_DAEMON_GENERATIONS.slack": "560f9edb3556499911a859bc2a1c3dc6dcf927da2674c46be409d5ee3cf46086",
"slack:packages/coding-agent/src/sdk/bus/chat-daemon-control.ts:ChatDaemonAction": "d8acaf90439e410595b5cd56fd187001393ea943517a5563a4a2ea3c88d155ef",
"slack:packages/coding-agent/src/sdk/bus/chat-daemon-control.ts:ChatDaemonController": "bb722a6a83eb383fce45e756cba356592b33ff0f505b4263409ce9138ea651f1",
"slack:packages/coding-agent/src/sdk/bus/chat-daemon-control.ts:ChatDaemonKind": "b1c2906c4eb04e120c9ce42f8b549a68cfc834d29d4d745335712a7f61bf09f2",
Expand Down Expand Up @@ -518,7 +518,7 @@
"slack:packages/coding-agent/src/sdk/bus/slack-daemon.ts:stop": "4ec3502bc5b1fc192a44a9f53f5d19d56190be51b5b285c171bc97441487dbf5",
"slack:packages/coding-agent/src/sdk/client/discovery.ts:readSdkSessionEndpoint": "bd8efa5f581993b3dc6c5549f62732211181b460b6e38d97220bb77b1875075e",
"slack:packages/coding-agent/src/sdk/router/session-router.ts:SessionRouter.#attach": "a08b206ed46659b003ff331708cfccdf81c5c3037263bacca71d675fd09321e3",
"slack:packages/coding-agent/src/sdk/router/session-router.ts:SessionRouter.#createAttachedClient": "76a3018aaf6f9b1dec802a884cb1e40c16a9c9b5ae5b842a70c1fc01c3cf53eb",
"slack:packages/coding-agent/src/sdk/router/session-router.ts:SessionRouter.#createAttachedClient": "f7d9879aeb8e84d4b4090567694b8332fdea95b25a79614e639a313dfd00fc87",
"slack:packages/coding-agent/src/sdk/router/session-router.ts:SessionRouter.#publishAttachment": "d46fff88a22ebd42e3e439a5cc20059a2ed002945972849e841a2e7d644a3079",
"telegram:packages/coding-agent/src/sdk/bus/config.ts:isTelegramSessionEligible": "6e2bf01ec7f58e07658172a2911741891f4eb05e2050e536bc170d6770746c37",
"telegram:packages/coding-agent/src/sdk/bus/config.ts:parseNotificationSettingsSnapshot": "a694158726a9a4f4df58d1b1a4cee54884ef2e4bab506b3c6cac981b3f41072e",
Expand Down
Loading