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
58 changes: 52 additions & 6 deletions apps/daemon/src/dispatch/daemonDispatcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import type { IConfigPresenter } from "@argos/shared/presenter";
import { resolveDaemonVersion } from "../version";
import type { DaemonTerminalRuntime } from "../terminal/daemonTerminalRuntime";
import { diagnoseDaemonSchema, repairDaemonSchema } from "../host/daemonSchemaDiagnostics";
import { settleSessionForOwnershipChange, type SettleSessionHost } from "../host/sessionSettlement";
import { getPiToolDefinitions } from "../host/piToolCatalog";
import { aggregateUsageStats, resolveBuiltinModelPrice } from "../host/usageStatsAggregator";
import { resolveModelCost } from "../host/modelCost";
Expand Down Expand Up @@ -371,6 +372,7 @@ type DaemonProviderExecutionPort = Required<
| "setAcpPreferredProcessMode"
| "prepareAcpSession"
| "clearAcpSession"
| "purgeAcpSessionData"
| "getAcpSessionModes"
| "setAcpSessionMode"
| "resolveAgentPermission"
Expand Down Expand Up @@ -939,6 +941,40 @@ export function createDaemonDispatcher(
sessionRepository: DaemonSessionRepositoryPort;
providerExecutionPort: DaemonProviderExecutionPort;
} = { sessionRepository, providerExecutionPort };
const settlementHost: SettleSessionHost = {
getSession: async (sessionId) => (await (sessionRepository as any).get?.(sessionId)) ?? null,
listPendingInputs: async (sessionId) => (await (sessionRepository as any).listPendingInputs?.(sessionId)) ?? [],
deletePendingInput: async (sessionId, itemId) => {
await (sessionRepository as any).deletePendingInput?.(sessionId, itemId);
},
cancelGeneration: (sessionId) => providerExecutionPort.cancelGeneration(sessionId),
purgeAcpSessionData: (sessionId) => providerExecutionPort.purgeAcpSessionData?.(sessionId),
};
/**
* Resolve the execution context a session receives when moved to `toAgentId`.
* ACP targets keep the historical `providerId: "acp"` + `modelId: <agent>`
* convention; Argos targets must receive the target agent's default model —
* labelling them `acp` would break the next send ("ACP agent not found").
*/
const resolveMoveTargetContext = async (
toAgentId: string,
): Promise<{ agentId: string; providerId: string; modelId: string }> => {
const agentType = await daemonConfig.getAgentType(toAgentId);
if (agentType === "acp") {
return { agentId: toAgentId, providerId: "acp", modelId: toAgentId };
}
if (agentType !== "argos") {
throw new Error(`Target agent not found: ${toAgentId}`);
}
const config = await daemonConfig.resolveArgosAgentConfig(toAgentId);
const defaultModel = daemonConfig.getDefaultModel();
const providerId = config?.defaultModelPreset?.providerId?.trim() || defaultModel?.providerId?.trim() || "";
const modelId = config?.defaultModelPreset?.modelId?.trim() || defaultModel?.modelId?.trim() || "";
if (!providerId || !modelId) {
throw new Error(`Target Argos agent does not have a default model: ${toAgentId}`);
}
return { agentId: toAgentId, providerId, modelId };
};
const daemonConfig = configPresenter as IConfigPresenter & DaemonMcpConfigPort & DaemonProviderConfigPort;
const daemonSettings = configPresenter as IConfigPresenter & DaemonScheduledTaskConfigPort;

Expand Down Expand Up @@ -3077,11 +3113,18 @@ export function createDaemonDispatcher(
if (route === sessionsMoveAgentSessionsRoute.name) {
const input = sessionsMoveAgentSessionsRoute.input.parse(rawInput);
const repo = runtime.sessionRepository as any;
// Validate the target before any destructive settlement: an invalid or
// model-less target must fail without discarding queued inputs or
// cancelling the source sessions.
const targetContext = await resolveMoveTargetContext(input.toAgentId);
const sessions = await repo.list({ agentId: input.fromAgentId, includeSubagents: true });
const movedSessionIds: string[] = [];
const deletedSessionIds: string[] = [];

for (const session of sessions) {
// Settle before the ownership change: discard queued inputs, cancel a
// running turn and wait for it to settle, release ACP bindings.
await settleSessionForOwnershipChange(session.id, settlementHost);
const messages = await repo.listMessages(session.id);
const children = await repo.list({ includeSubagents: true, parentSessionId: session.id });
const isEmptyDraft = Boolean(session.isDraft) && messages.length === 0 && children.length === 0;
Expand All @@ -3091,9 +3134,7 @@ export function createDaemonDispatcher(
continue;
}
await repo.moveSessionToAgent(session.id, {
agentId: input.toAgentId,
providerId: "acp",
modelId: input.toAgentId,
...targetContext,
projectDir: session.projectDir ?? null,
permissionMode: session.permissionMode ?? "default",
subagentEnabled: Boolean(session.subagentEnabled),
Expand All @@ -3115,6 +3156,9 @@ export function createDaemonDispatcher(
const sessions = await repo.list({ agentId: input.agentId, includeSubagents: true });
const deletedSessionIds: string[] = [];
for (const session of sessions) {
// Settle first: cancel a running turn (bounded wait) and discard
// queued inputs so deletion cannot race the runtime.
await settleSessionForOwnershipChange(session.id, settlementHost);
await repo.delete(session.id);
deletedSessionIds.push(session.id);
}
Expand All @@ -3128,10 +3172,11 @@ export function createDaemonDispatcher(
if (!session) {
throw new Error(`Session not found: ${input.sessionId}`);
}
// Validate the target before the destructive settlement.
const targetContext = await resolveMoveTargetContext(input.toAgentId);
await settleSessionForOwnershipChange(input.sessionId, settlementHost);
const updated = await repo.moveSessionToAgent(input.sessionId, {
agentId: input.toAgentId,
providerId: "acp",
modelId: input.toAgentId,
...targetContext,
projectDir: session.projectDir ?? null,
permissionMode: session.permissionMode ?? "default",
subagentEnabled: Boolean(session.subagentEnabled),
Expand All @@ -3143,6 +3188,7 @@ export function createDaemonDispatcher(

if (route === sessionsDeleteRoute.name) {
const input = sessionsDeleteRoute.input.parse(rawInput);
await settleSessionForOwnershipChange(input.sessionId, settlementHost);
await (runtime as any).sessionRepository.delete(input.sessionId);
return sessionsDeleteRoute.output.parse({ deleted: true });
}
Expand Down
10 changes: 10 additions & 0 deletions apps/daemon/src/host/acp-provider-execution.ts
Original file line number Diff line number Diff line change
Expand Up @@ -881,6 +881,16 @@ export class AcpProviderExecutionPort implements ProviderExecutionPort {
}
}

async purgeAcpSessionData(sessionId: string): Promise<void> {
// Best-effort: stop any active turn first so the binding cannot be
// re-created mid-purge, then delete the durable `acp_sessions` rows.
// Row deletion failures propagate: a binding left behind would keep the
// ACP uninstall guard stuck permanently.
await this.cancelGeneration(sessionId).catch(() => undefined);
const runtime = await this.getRuntime();
await runtime.sessionPersistence.deleteAllSessions(sessionId);
}

async respondToolInteraction(
sessionId: string,
_messageId: string,
Expand Down
20 changes: 20 additions & 0 deletions apps/daemon/src/host/daemonAcpConfig.ts
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,26 @@ export class DaemonAcpConfig {
return this.acpConfHelper.getGlobalEnabled();
}

/**
* Resolve the ACP agent type regardless of enabled/install state. Unlike
* `getAcpAgents()` (which only surfaces enabled+installed agents), this lets
* callers identify sessions bound to a disabled or uninstalled agent so they
* can be moved or deleted before agent removal.
*/
getAcpAgentTypeIncludingState(agentId: string): "acp" | null {
const resolvedId = resolveAcpAgentAlias(agentId);
const manual = this.acpConfHelper.getManualAgents().some((agent) => agent.id === resolvedId);
if (manual) return "acp";
try {
const registered = this.acpRegistryService.listAgents().some((agent) => agent.id === resolvedId);
return registered ? "acp" : null;
} catch (error) {
// A missing/unreadable registry snapshot must not break type resolution.
logger.warn("[ACP] registry agent lookup failed:", error);
return null;
}
}

async setAcpEnabled(enabled: boolean): Promise<void> {
this.acpConfHelper.setGlobalEnabled(enabled);
}
Expand Down
17 changes: 16 additions & 1 deletion apps/daemon/src/host/daemonConfigPresenter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,8 @@ import { readFileSync, renameSync, writeFileSync, existsSync, mkdirSync } from "
import { homedir } from "node:os";
import { join, dirname } from "node:path";
import { fileURLToPath } from "node:url";
import { DEFAULT_PROVIDERS, normalizeScheduledTasksConfig } from "@argos/backend-core";
import { DEFAULT_PROVIDERS, normalizeScheduledTasksConfig, resolveAcpAgentAlias } from "@argos/backend-core";
import { BUILTIN_ARGOS_AGENT_ID } from "@argos/agent-runtime";
import { ProviderDbLoader, resolveAiSdkProviderDefinition } from "@argos/backend-core/provider";
import type {
BuiltinKnowledgeConfig,
Expand Down Expand Up @@ -1264,6 +1265,20 @@ export class DaemonConfigPresenter {
return this.argosAgentRuntime ? this.argosAgentRuntime.getAgent(agentId) : null;
}

/**
* State-agnostic agent-type resolution. Registry/manual ACP agents resolve
* even when disabled or uninstalled so sessions bound to them remain
* movable/deletable ahead of agent removal (see
* docs/issues/acp-agent-removal-settlement).
*/
async getAgentType(agentId: string): Promise<"argos" | "acp" | null> {
const resolvedId = resolveAcpAgentAlias(String(agentId ?? "").trim());
if (!resolvedId) return null;
if (resolvedId === BUILTIN_ARGOS_AGENT_ID) return "argos";
if (this.argosAgentRuntime?.getAgent(resolvedId)) return "argos";
return this.acpConfig.getAcpAgentTypeIncludingState(resolvedId);
}

async getArgosAgentConfig(agentId: string): Promise<any> {
return this.argosAgentRuntime ? this.argosAgentRuntime.getArgosAgentConfig(agentId) : null;
}
Expand Down
105 changes: 105 additions & 0 deletions apps/daemon/src/host/sessionSettlement.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,105 @@
import type { PendingSessionInputRecord } from "@argos/shared/types/agent-interface";

/**
* Settlement for session ownership changes (delete / move / agent removal).
*
* Mirrors the upstream DeepChat fix for "allow uninstall while disabled"
* (ThinkInAIXYZ/deepchat#2188), re-implemented natively for Argos' daemon-owned
* session architecture:
*
* 1. Discard queue-mode pending inputs — they belong to no turn yet and would
* otherwise leak onto the next owner. Steer-mode inputs are kept on purpose:
* they are conversation facts, and a cancelled run cannot claim them because
* `cancelGeneration` suppresses the pending-input drain.
* 2. Cancel an active generation and wait (bounded) for the session status to
* leave `generating` — cancellation settles asynchronously, so proceeding
* immediately would race the runtime.
* 3. Purge durable ACP bindings (`acp_sessions` rows) best-effort so the ACP
* uninstall guard becomes accurate after the sessions are gone.
*/

export interface SettleSessionHost {
getSession(sessionId: string): Promise<{ status?: string | null } | null>;
listPendingInputs(sessionId: string): Promise<PendingSessionInputRecord[]>;
deletePendingInput(sessionId: string, itemId: string): Promise<void>;
cancelGeneration(sessionId: string): Promise<void>;
purgeAcpSessionData?(sessionId: string): Promise<void>;
}

export interface SettleSessionOptions {
/** Max time to wait for a cancelled generation to settle. Default 10s. */
timeoutMs?: number;
/** Poll interval while waiting for settlement. Default 100ms. */
pollIntervalMs?: number;
/** Injectable delay for tests. */
delay?: (ms: number) => Promise<void>;
}

export interface SettleSessionResult {
cancelled: boolean;
discardedQueueInputIds: string[];
}

const DEFAULT_TIMEOUT_MS = 10_000;
const DEFAULT_POLL_INTERVAL_MS = 100;

export async function settleSessionForOwnershipChange(
sessionId: string,
host: SettleSessionHost,
options: SettleSessionOptions = {},
): Promise<SettleSessionResult> {
const delay = options.delay ?? ((ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)));
const discardedQueueInputIds = await discardQueueInputs(sessionId, host);

let cancelled = false;
if ((await currentStatus(sessionId, host)) === "generating") {
cancelled = true;
await host.cancelGeneration(sessionId);
await waitForSettle(sessionId, host, delay, options);
}

// Fail-closed: a durable binding that cannot be purged would leave the ACP
// uninstall guard stuck with no conversation left to retry against, so the
// ownership change must abort instead of proceeding.
await host.purgeAcpSessionData?.(sessionId);

return { cancelled, discardedQueueInputIds };
}

async function currentStatus(sessionId: string, host: SettleSessionHost): Promise<string | null> {
// Propagate read failures: treating a transient error as "not generating"
// could skip cancellation and race a still-running turn.
const session = await host.getSession(sessionId);
return session?.status ?? null;
}

async function discardQueueInputs(sessionId: string, host: SettleSessionHost): Promise<string[]> {
// Fail-closed: if queued inputs cannot be enumerated, the ownership change
// must stop — leaked inputs would drain under the new owner.
const inputs = await host.listPendingInputs(sessionId);
const discarded: string[] = [];
for (const input of inputs) {
if (input.mode !== "queue") continue;
await host.deletePendingInput(sessionId, input.id);
discarded.push(input.id);
}
Comment on lines +76 to +85

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Failed Cleanup Leaks Inputs

Failures to list or delete queued inputs are swallowed, so a move continues as though cleanup succeeded. Moving a session preserves its pending-input rows; therefore, any queue input left behind remains attached to the same session ID and can be consumed by the newly assigned agent even though it was intended for the previous agent. The move should stop when queued inputs cannot be enumerated or removed.

Prompt To Fix With AI
This is a comment left during a code review.
Path: apps/daemon/src/host/sessionSettlement.ts
Line: 75-87

Comment:
**Failed Cleanup Leaks Inputs**

Failures to list or delete queued inputs are swallowed, so a move continues as though cleanup succeeded. Moving a session preserves its pending-input rows; therefore, any queue input left behind remains attached to the same session ID and can be consumed by the newly assigned agent even though it was intended for the previous agent. The move should stop when queued inputs cannot be enumerated or removed.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Fix in Codex

return discarded;
}

async function waitForSettle(
sessionId: string,
host: SettleSessionHost,
delay: (ms: number) => Promise<void>,
options: SettleSessionOptions,
): Promise<void> {
const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
const pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS;
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
await delay(pollIntervalMs);
if ((await currentStatus(sessionId, host)) !== "generating") {
return;
}
}
throw new Error(`Session ${sessionId} did not stop before ownership change.`);
}
5 changes: 5 additions & 0 deletions apps/daemon/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ type DaemonProviderExecutionPort = Required<
| "setAcpPreferredProcessMode"
| "prepareAcpSession"
| "clearAcpSession"
| "purgeAcpSessionData"
| "getAcpSessionModes"
| "setAcpSessionMode"
| "resolveAgentPermission"
Expand Down Expand Up @@ -457,6 +458,10 @@ export async function startDaemon(options?: {
async clearAcpSession(sessionId) {
return acpProviderExecutionPort.clearAcpSession(sessionId);
},
async purgeAcpSessionData(sessionId) {
// ACP-only: pi sessions carry no durable binding rows.
return acpProviderExecutionPort.purgeAcpSessionData(sessionId);
},
async getAcpSessionModes(conversationId) {
return acpProviderExecutionPort.getAcpSessionModes(conversationId);
},
Expand Down
58 changes: 58 additions & 0 deletions apps/daemon/test/daemonAcpConfig.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -261,4 +261,62 @@ describe("DaemonAcpConfig reconcileInstalledAgents", () => {

await expect(config.uninstallAcpRegistryAgent(binaryOkAgent.id)).rejects.toThrow("still has related conversations");
});

describe("getAcpAgentTypeIncludingState", () => {
it("resolves disabled and uninstalled registry agents", async () => {
const harness = createConfig({
// disabled-agent is registered but disabled; locked-agent has no
// install state at all (never installed).
registryStates: { [disabledAgent.id]: { enabled: false } },
});

await harness.config.initialReconcile;

expect(harness.config.getAcpAgentTypeIncludingState(disabledAgent.id)).toBe("acp");
expect(harness.config.getAcpAgentTypeIncludingState(binaryFailAgent.id)).toBe("acp");
expect(harness.config.getAcpAgentTypeIncludingState("unknown-agent")).toBeNull();
});

it("resolves disabled manual agents and tolerates alias lookups", async () => {
const configDir = fs.mkdtempSync(path.join(os.tmpdir(), "argos-acp-cfg-"));
const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), "argos-acp-data-"));
roots.push(configDir, dataDir);

fs.mkdirSync(path.join(dataDir, "acp-registry"), { recursive: true });
fs.writeFileSync(
path.join(dataDir, "acp-registry", "meta.json"),
JSON.stringify({ version: "1.0.0", lastUpdated: Date.now(), lastAttemptedAt: Date.now(), sourceUrl: "" }),
);
fs.writeFileSync(
path.join(dataDir, "acp-registry", "registry.json"),
JSON.stringify({ version: "1.0.0", agents: [] }),
);
fs.mkdirSync(configDir, { recursive: true });
fs.writeFileSync(
path.join(configDir, "acp_agents.json"),
JSON.stringify({
enabled: true,
version: "4",
registryStates: {},
manualAgents: [
{
id: "manual-1",
name: "Manual Agent",
source: "manual",
enabled: false,
command: "manual-agent",
args: [],
},
],
installStates: {},
sharedMcpSelections: [],
}),
);

const config = new DaemonAcpConfig({ configDir, dataDir });

expect(config.getAcpAgentTypeIncludingState("manual-1")).toBe("acp");
expect(config.getAcpAgentTypeIncludingState("unknown")).toBeNull();
});
});
});
Loading
Loading