mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-07-09 17:19:02 +00:00
fix(daemon): resolve ACP permission votes across connections (#5912)
* fix(daemon): resolve acp permission votes across connections * fix(daemon): address review on acp permission vote resolution Resolve the review feedback on PR #5912: - dispatch.ts session/permission: add server-side stderr logging to every failure mode (missing requestId, no pending entry, ownership failure, bridge rejection) so a stuck permission prompt is debuggable, matching the legacy resolveClientResponse path. - On bridge rejection (accepted === false), stop deleting the pending entry and stop reusing the "no pending" error. Keep the entry until teardown (as the legacy path does) and return a distinct 409 "vote not accepted" error, so the two states aren't conflated and a retry on another connection can still land. - connection-registry.ts: extract findPendingPermissionEntry shared by findPendingPermission and deletePendingPermission so the matching predicate lives in one place; delete now stops at the first (globally unique) match. - index.ts: the abandonPending callback logs-and-returns-false before the dispatcher is initialized instead of throwing through the teardown path, matching the detachClient callback's defensive posture. - Tests: cover the previously-untested handler branches (missing requestId, invalid outcome shapes, cancelled outcome, bridge rejection + sessionId inference + entry retention) and assert the connection-qualified id format and the undefined-sessionId lookup branch. Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(daemon): log dropped cross-connection vote + cover handler error branches Address the second ci-bot review round on PR #5912: - dispatch.ts resolveClientResponse: the cross-connection ownership guard dropped a vote silently. Add a writeStderrLine so a vote rejected on the legacy path leaves the same grep-friendly operator signal the session/permission handler already emits — otherwise the agent's prompt stays blocked until teardown with no log to correlate. - transport.test.ts: add end-to-end coverage for two previously-untested handler branches — the no-pending 404 response (requestId misses the registry with no sessionId) and the unowned-session rejection (a connection voting on a session it does not own). Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * refactor(daemon): harden acp permission lookup and align registry api Address the third ci-bot review round on PR #5912 (all non-blocking suggestions): - connection-registry.ts: collapse the redundant private findPendingPermissionEntry pass-through into the public findPendingPermission, and align deletePendingPermission to the same (requestId, sessionId) argument order so the two can never be called with swapped string args (a swap would silently match nothing and leak the entry until teardown, with no type error). - dispatch.ts session/permission: look the pending entry up by the globally-unique requestId alone and treat the entry's own session as authoritative; when the client supplies a sessionId that does not match, reject with an explicit 409 instead of routing requireOwned and the bridge vote at the wrong session (which left the real entry to leak until teardown). - Tests: add the sessionId-mismatch rejection case and update call sites for the new argument order. Out of scope and deferred: making the dispatcher's registry a required constructor parameter (and the dependent dropResolvedPermission cleanup) — that changes the constructor contract beyond this fix. Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(daemon): delete resolved permission by precise map key, fix comment Address wenshao's review on PR #5912: findPendingPermission matches on bridgeRequestId (a per-request randomUUID), not the connection-qualified conn.pending map key. Under multi-client attach a permission_request reaches every co-owning connection, each minting its own entry that shares the same bridgeRequestId — so more than one entry can match and the prior "globally unique, at most one match" comment was wrong. - connection-registry.ts: correct the findPendingPermission doc to attribute uniqueness to the map key (not matched here) and note co-owning connections can share a bridgeRequestId, so callers needing a specific entry must act on the conn/map-key they already hold. - dispatch.ts dropResolvedPermission: delete the resolved entry by its exact conn/map-key instead of re-matching by bridgeRequestId, which under multi-attach could delete a sibling connection's entry and orphan the one just resolved. Drops the now-unused req parameter. deletePendingPermission stays for the session/permission handler, where the lookup and delete consistently target the same first match. Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(daemon): align acp permission errors with REST 404/400/403 shapes Address wenshao's two [Critical] review findings on PR #5912: - session/permission no longer falls through to the bridge when the registry misses. In the scoped route a sessionId is always supplied, so a stale/ unknown requestId previously routed to the caller's session, got a bridge `false`, and was reported as a thrown 409 — diverging from the established `404 -> false` contract of DaemonClient.respondToSessionPermission() and the REST route. Now a registry miss returns 404; 409 is reserved for a present entry the bridge still rejects. - Wrap the bridge vote and map permission-specific throws like REST's sendPermissionVoteError: InvalidPermissionOptionError -> INVALID_PARAMS with httpStatus 400 + invalid_option_id, PermissionForbiddenError -> httpStatus 403 + permission_forbidden (with requestId/sessionId/reason). Previously these fell through the outer catch into a generic httpStatus-less internal error, so SDK callers saw 500s for normal permission outcomes. Import the error classes from acp-session-bridge (as REST does) so instanceof matches the class the bridge throws. - Tests: cover the 404-on-miss-with-sessionId regression and the 400/403 mappings. Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * refactor(daemon): unify permission delete, whitelist vote forward Address the fourth ci-bot review round on PR #5912 (non-blocking suggestions): - session/permission success path now drops the resolved entry through the shared dropResolvedPermission helper using the conn/map-key pendingRef already carries, instead of re-matching by requestId. Unifies the two delete sites and keeps the deletion precise. - parsePermissionResponse forwards only the bridge-contract fields (outcome plus the ACP-reserved _meta passthrough) rather than copying every remaining client key, removing a needless client-controlled surface on the server-side bridge argument. - transport.test.ts: the cross-connection permission test now asserts a duplicate vote on the same id does not reach the bridge again, locking down the cleanup guarantee that is the core of this PR. Declined (replied on the threads): a blanket local try/catch around the bridge vote (would shadow the outer dispatcher's typed-error mapping for non-permission errors) and success-side audit logging in the generic findPendingClientRequest (log noise / out of scope). Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(daemon): full REST parity for acp permission vote errors Address the fifth ci-bot review round on PR #5912: - session/permission now maps every permission-specific bridge throw like REST's sendPermissionVoteError: InvalidPermissionOptionError -> 400, PermissionForbiddenError -> 403, PermissionPolicyNotImplementedError -> 501 (policy), CancelSentinelCollisionError -> 500 (requestId/sentinel). The last two previously fell through to the outer dispatcher catch and became a generic -32603 without structured metadata. - Truly unexpected bridge/sessionCtx failures now run the same cancelAbandonedPermission fallback as the legacy resolveClientResponse path (dropping the entry only if the cancel landed, else keeping it for teardown) before rethrowing — so an unexpected error no longer leaves the mediator blocking the agent's prompt until session teardown. - parsePermissionResponse rebuilds the outcome sub-object from its validated keys instead of forwarding it verbatim, so a client can no longer inject extra outcome sub-fields (e.g. force) into the bridge argument; _meta is forwarded only when it is an object. - Tests: cross-connection vote via the session/permission method (ack on the voter's stream + entry removed from the originator), plus the 501 and 500 error mappings. Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(daemon): preserve answers payload and delete only the voter's own entry Address wenshao's two [Critical] findings on PR #5912: - parsePermissionResponse no longer drops the AskUserQuestion `answers` payload. The whitelist tightening forwarded only outcome/_meta, but the bridge treats `answers` (an object map of string values) as the one supported non-ACP permission-response field, so votes were resolving while the agent received no submitted answers. Forward it under the same shape the bridge validates. - The session/permission success path now deletes only the voting connection's OWN pending entry for the requestId, not the first registry-wide match. pendingRef can belong to a sibling connection; under the consensus policy respondToSessionPermission returns true for an intermediate "recorded" vote, so deleting a sibling's entry could drop a co-owner's still-needed request and stall the quorum. A cross-connection voter with no own entry deletes nothing and leaves the originator's entry for teardown. - Tests: forward-answers/strip-unknown-fields case, and the cross-connection method test now asserts a co-owner's vote does NOT delete the originator's sibling entry. Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(daemon): validate legacy vote path and scope permission deletes to voter Address the sixth ci-bot review round on PR #5912 (2 Critical + 4 suggestions): - resolveClientResponse now validates/whitelists the client result through the same parsePermissionResponse the session/permission handler uses. This PR had widened that legacy path to any co-owning connection (via findPendingClientRequest), so the raw `result as unknown` cast was a cross-connection injection surface for arbitrary top-level args and extra outcome sub-fields; a malformed result still throws and hits the cancel fallback as before. - The unexpected-error cancel fallback in the session/permission handler now drops only the VOTING connection's own entry (via the new shared dropOwnPendingPermission helper), not pendingRef — which is the first registry-wide match and may be the originator's entry, whose deletion would stall a consensus quorum still awaiting other co-owners. - parsePermissionResponse logs a stderr line when a present-but-malformed `answers` is dropped, instead of silently discarding it. - Removed ConnectionRegistry.deletePendingPermission: it had no production callers and its first-match semantics were unsafe under co-owned sessions (deletion is done connection-scoped in the dispatcher). - Tests: _meta object-preserved / non-object-dropped, and the generic unexpected-error fallthrough (cancel fallback runs + error propagates). Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(daemon): structured 403 for unowned session/permission vote Address wenshao's remaining [Critical] on PR #5912: the session/permission ownership rejection went through the shared requireOwned, which sends an INVALID_PARAMS error with no `data` envelope. Every other error path in this handler carries `{ httpStatus }` (404/409/400/403/500/501), so SDK callers that classify permission-vote failures by error.data.httpStatus got undefined for the likeliest cross-connection failure (right session header, no session/new on this connection). Inline the ownership check so the rejection carries httpStatus 403 + sessionId + requestId, leaving the shared requireOwned untouched for other handlers. Test asserts the 403. (wenshao's other two criticals — legacy raw-result forwarding and the catch-all deleting the originator's entry — were already fixed in 4bb06e5fd.) Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(daemon): structured 400 for invalid session/permission params Address the latest ci-bot suggestion on PR #5912: parsePermissionResponse throws AcpParamError, a plain Error with no httpStatus, which the outer dispatcher catch maps to a bare INVALID_PARAMS — inconsistent with every other error path in this handler (404/409/400/403/500/501 all carry httpStatus). Catch AcpParamError locally and return a structured 400 with requestId, so SDK callers that classify by error.data.httpStatus see a consistent shape. The parametrized invalid-outcome test now asserts the 400. Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * perf(daemon): O(1) pending lookup, log vote success, cover malformed answers Address wenshao's three suggestions on PR #5912: - findPendingClientRequest parses the originating connectionId from the server-minted id format (_qwen_perm_<connectionId>_<counter>) for an O(1) byId lookup, falling back to the full scan for client-chosen ids. - The session/permission success path now writes a stderr line ("vote accepted") so an operator debugging a stuck prompt can tell it apart from "vote never arrived" or "landed on another connection" — every failure branch already logs. - Added a test for the malformed-answers branch (non-string values) asserting the vote still lands but answers are not forwarded to the bridge. Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(daemon): httpStatus on missing-requestId + cross-conn malformed vote test Address the latest ci-bot review on PR #5912: - The missing-`requestId` rejection in session/permission was the only error branch without an { httpStatus } envelope. Add httpStatus 400 (+ requestId) so SDK callers can classify it like every other validation error here. Test asserts the 400. - Add an integration test for the legacy resolveClientResponse cross-connection variant: connection B (a co-owner) answers connection A's permission request with a malformed result, parsePermissionResponse (added for this path) throws, and the cancel fallback still releases the mediator. Generated with AI Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> --------- Co-authored-by: 秦奇 <gary.gq@alibaba-inc.com> Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
ChiGao
GitHub
parent
7b9e31885b
commit
c48dc32b57
5 changed files with 1812 additions and 29 deletions
|
|
@ -859,4 +859,49 @@ describe('ConnectionRegistry.getSnapshot', () => {
|
|||
registry.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it('finds pending permissions across connections', () => {
|
||||
const registry = new ConnectionRegistry();
|
||||
try {
|
||||
const connA = registry.create(false);
|
||||
const connB = registry.create(false);
|
||||
expect(connA).toBeDefined();
|
||||
expect(connB).toBeDefined();
|
||||
if (!connA || !connB) return;
|
||||
|
||||
const idA = connA.nextId();
|
||||
const idB = connB.nextId();
|
||||
expect(idA).not.toBe(idB);
|
||||
// Pin the connection-qualified format `_qwen_perm_<connectionId>_<N>` —
|
||||
// it's the collision-prevention guarantee of this change, so a
|
||||
// regression to the old `_qwen_perm_<N>` format must fail here, not just
|
||||
// an "ids differ" check that the old format would also pass.
|
||||
expect(idA).toMatch(/^_qwen_perm_.+_1$/);
|
||||
expect(idA).toContain(connA.connectionId);
|
||||
expect(idB).toContain(connB.connectionId);
|
||||
|
||||
connA.pending.set(idA, {
|
||||
sessionId: 'sess-1',
|
||||
bridgeRequestId: 'perm-1',
|
||||
kind: 'permission',
|
||||
});
|
||||
|
||||
expect(registry.findPendingClientRequest(idA)?.conn).toBe(connA);
|
||||
expect(registry.findPendingPermission('perm-1', 'sess-1')?.id).toBe(idA);
|
||||
// The `sessionId === undefined` branch (relied on by dispatch's
|
||||
// `session/permission` handler when no sessionId is supplied) matches on
|
||||
// requestId alone, while a mismatched sessionId must not match.
|
||||
expect(registry.findPendingPermission('perm-1')?.id).toBe(idA);
|
||||
expect(
|
||||
registry.findPendingPermission('perm-1', 'wrong-session'),
|
||||
).toBeUndefined();
|
||||
|
||||
// findPendingPermission is a read-only locator; deletion is done by the
|
||||
// owning connection on its own map key (AcpDispatcher.dropOwnPendingPermission).
|
||||
connA.pending.delete(idA);
|
||||
expect(registry.findPendingClientRequest(idA)).toBeUndefined();
|
||||
} finally {
|
||||
registry.dispose();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -145,6 +145,12 @@ export interface PendingClientRequest {
|
|||
kind: 'permission';
|
||||
}
|
||||
|
||||
export interface PendingClientRequestRef {
|
||||
conn: AcpConnection;
|
||||
id: string;
|
||||
req: PendingClientRequest;
|
||||
}
|
||||
|
||||
export interface AcpConnectionDiagnostic {
|
||||
connectionIdPrefix: string;
|
||||
fromLoopback: boolean;
|
||||
|
|
@ -254,13 +260,13 @@ export class AcpConnection {
|
|||
|
||||
/**
|
||||
* Allocate a fresh JSON-RPC id for an agent→client request. STRING-typed
|
||||
* (`_qwen_perm_N`) so it can never collide with a client-originated id —
|
||||
* (`_qwen_perm_<conn>_N`) so it can never collide with a client-originated id —
|
||||
* JSON-RPC 2.0 permits clients to use any number (incl. negatives) or
|
||||
* string, so a numeric namespace wasn't actually safe.
|
||||
*/
|
||||
nextId(): string {
|
||||
this.idCounter += 1;
|
||||
return `_qwen_perm_${this.idCounter}`;
|
||||
return `_qwen_perm_${this.connectionId}_${this.idCounter}`;
|
||||
}
|
||||
|
||||
touch(): void {
|
||||
|
|
@ -901,6 +907,64 @@ export class ConnectionRegistry {
|
|||
return conn;
|
||||
}
|
||||
|
||||
findPendingClientRequest(id: string): PendingClientRequestRef | undefined {
|
||||
// Fast path: server-minted ids embed their originating connection
|
||||
// (`_qwen_perm_<connectionId>_<counter>`, connectionId is a hyphenated
|
||||
// `randomUUID()` with no underscores), so the owning connection is the
|
||||
// substring before the last `_` — an O(1) lookup instead of scanning all
|
||||
// connections on every client response.
|
||||
if (id.startsWith('_qwen_perm_')) {
|
||||
const body = id.slice('_qwen_perm_'.length);
|
||||
const lastUnderscore = body.lastIndexOf('_');
|
||||
if (lastUnderscore > 0) {
|
||||
const conn = this.byId.get(body.slice(0, lastUnderscore));
|
||||
const req = conn?.pending.get(id);
|
||||
if (conn && req) return { conn, id, req };
|
||||
}
|
||||
}
|
||||
// Fallback for client-chosen ids that don't match the server format.
|
||||
for (const conn of this.byId.values()) {
|
||||
const req = conn.pending.get(id);
|
||||
if (req) return { conn, id, req };
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Locate a pending permission entry matching `requestId` (a bridge
|
||||
* `bridgeRequestId`, i.e. a per-request `randomUUID()`) and optionally
|
||||
* `sessionId`, returning the first match.
|
||||
*
|
||||
* NOTE: `requestId` is unique per *request*, not per *pending entry*. The
|
||||
* per-entry unique id is the `conn.pending` map key
|
||||
* (`_qwen_perm_<connectionId>_N`), which is NOT what is matched here. A
|
||||
* `permission_request` is delivered to every live subscriber of its session,
|
||||
* so when connections co-own a session (multi-client attach) each mints its
|
||||
* own entry sharing the same `bridgeRequestId`. More than one entry can
|
||||
* therefore match, so this is a *read-only locator* for deriving a session /
|
||||
* ownership from a wire `requestId`. To DELETE a resolved entry, act on the
|
||||
* specific `conn`/map-key the caller already holds (see
|
||||
* `AcpDispatcher.dropOwnPendingPermission`) — never delete by re-matching
|
||||
* here, which could hit a sibling co-owner's entry.
|
||||
*/
|
||||
findPendingPermission(
|
||||
requestId: string,
|
||||
sessionId?: string,
|
||||
): PendingClientRequestRef | undefined {
|
||||
for (const conn of this.byId.values()) {
|
||||
for (const [id, req] of conn.pending) {
|
||||
if (
|
||||
req.kind === 'permission' &&
|
||||
req.bridgeRequestId === requestId &&
|
||||
(sessionId === undefined || req.sessionId === sessionId)
|
||||
) {
|
||||
return { conn, id, req };
|
||||
}
|
||||
}
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
delete(connectionId: string): boolean {
|
||||
const conn = this.byId.get(connectionId);
|
||||
if (!conn) return false;
|
||||
|
|
|
|||
|
|
@ -17,6 +17,15 @@ import {
|
|||
writeWorkspaceContextFile,
|
||||
type SubagentLevel,
|
||||
} from '@qwen-code/qwen-code-core';
|
||||
// Import the permission error classes from the same module REST's
|
||||
// `sendPermissionVoteError` uses, so `instanceof` matches the class the bridge
|
||||
// actually throws (the core re-export is a distinct identity at runtime).
|
||||
import {
|
||||
CancelSentinelCollisionError,
|
||||
InvalidPermissionOptionError,
|
||||
PermissionForbiddenError,
|
||||
PermissionPolicyNotImplementedError,
|
||||
} from '../acp-session-bridge.js';
|
||||
import { FsError } from '../fs/errors.js';
|
||||
import {
|
||||
TooManyActiveDeviceFlowsError,
|
||||
|
|
@ -72,7 +81,11 @@ import {
|
|||
WorkspacePermissionRulesSessionRequiredError,
|
||||
WorkspaceSettingsPartialPersistError,
|
||||
} from '../workspace-service/types.js';
|
||||
import type { AcpConnection } from './connection-registry.js';
|
||||
import type {
|
||||
AcpConnection,
|
||||
ConnectionRegistry,
|
||||
PendingClientRequestRef,
|
||||
} from './connection-registry.js';
|
||||
import {
|
||||
QWEN_META_KEY,
|
||||
QWEN_METHOD_NS,
|
||||
|
|
@ -96,7 +109,13 @@ function errMsg(err: unknown): string {
|
|||
return err instanceof Error ? err.message : String(err);
|
||||
}
|
||||
|
||||
type PermissionResponse = Parameters<
|
||||
HttpAcpBridge['respondToSessionPermission']
|
||||
>[2];
|
||||
|
||||
const SESSION_SHELL_METHOD = `${QWEN_METHOD_NS}session/shell`;
|
||||
const INVALID_PERMISSION_OUTCOME_ERROR =
|
||||
'`outcome` must be `{ outcome: "cancelled" }` or `{ outcome: "selected", optionId: string }`';
|
||||
|
||||
const ALL_QWEN_VENDOR_METHODS: readonly string[] = [
|
||||
`${QWEN_METHOD_NS}session/heartbeat`,
|
||||
|
|
@ -178,6 +197,7 @@ const CONN_ROUTED_METHODS = new Set<string>([
|
|||
'session/list',
|
||||
'session/close',
|
||||
'session/fork',
|
||||
'session/permission',
|
||||
...ALL_QWEN_VENDOR_METHODS,
|
||||
]);
|
||||
|
||||
|
|
@ -275,6 +295,61 @@ function validatePrompt(params: Record<string, unknown>): void {
|
|||
}
|
||||
}
|
||||
|
||||
function parsePermissionResponse(
|
||||
params: Record<string, unknown>,
|
||||
): PermissionResponse {
|
||||
const outcome = params['outcome'];
|
||||
if (!isObject(outcome)) {
|
||||
throw new AcpParamError(INVALID_PERMISSION_OUTCOME_ERROR);
|
||||
}
|
||||
if (outcome['outcome'] !== 'cancelled') {
|
||||
if (
|
||||
outcome['outcome'] !== 'selected' ||
|
||||
typeof outcome['optionId'] !== 'string' ||
|
||||
outcome['optionId'].length === 0
|
||||
) {
|
||||
throw new AcpParamError(INVALID_PERMISSION_OUTCOME_ERROR);
|
||||
}
|
||||
}
|
||||
|
||||
// Whitelist only the fields the bridge contract defines — a rebuilt `outcome`
|
||||
// carrying just its validated keys, plus the ACP-reserved `_meta` passthrough
|
||||
// — rather than forwarding client-supplied keys. The previous copy-all let a
|
||||
// client inject arbitrary top-level args AND extra `outcome` sub-fields (e.g.
|
||||
// `{ outcome: { outcome: 'selected', optionId: 'allow', force: true } }`) into
|
||||
// the server-side bridge call; harmless today since the bridge reads only the
|
||||
// discriminant + optionId, but a needless client-controlled surface.
|
||||
const cleanOutcome =
|
||||
outcome['outcome'] === 'cancelled'
|
||||
? { outcome: 'cancelled' as const }
|
||||
: { outcome: 'selected' as const, optionId: outcome['optionId'] };
|
||||
const response: Record<string, unknown> = { outcome: cleanOutcome };
|
||||
// `answers` is the one non-ACP permission-response field the bridge honours
|
||||
// (AskUserQuestion). Forward it under the same shape the bridge validates —
|
||||
// an object map of string values — so dropping it doesn't silently leave the
|
||||
// agent with no submitted answers.
|
||||
const answers = params['answers'];
|
||||
if (answers !== undefined) {
|
||||
if (
|
||||
isObject(answers) &&
|
||||
Object.values(answers).every((value) => typeof value === 'string')
|
||||
) {
|
||||
response['answers'] = answers;
|
||||
} else {
|
||||
// Present but malformed: the bridge would reject it anyway, so it is
|
||||
// dropped — but log it, otherwise a client whose answers silently
|
||||
// vanished (agent sees none) has no server-side signal to chase.
|
||||
writeStderrLine(
|
||||
'qwen serve: /acp session/permission dropping invalid `answers` (expected an object map of string values)',
|
||||
);
|
||||
}
|
||||
}
|
||||
if (isObject(params['_meta'])) {
|
||||
response['_meta'] = params['_meta'];
|
||||
}
|
||||
return response as PermissionResponse;
|
||||
}
|
||||
|
||||
/**
|
||||
* Map a thrown error to a JSON-RPC error code + a client-safe message.
|
||||
* Param-validation errors are echoed (they describe the client's own bad
|
||||
|
|
@ -431,6 +506,7 @@ export class AcpDispatcher {
|
|||
private readonly fsFactory?: WorkspaceFileSystemFactory,
|
||||
private readonly deviceFlowRegistry?: DeviceFlowRegistry,
|
||||
private readonly sessionShellCommandEnabled: boolean = false,
|
||||
private readonly registry?: ConnectionRegistry,
|
||||
) {
|
||||
this.agentManager = createDaemonSubagentManager(boundWorkspace);
|
||||
}
|
||||
|
|
@ -673,6 +749,44 @@ export class AcpDispatcher {
|
|||
return false;
|
||||
}
|
||||
|
||||
private findPendingClientRequest(
|
||||
conn: AcpConnection,
|
||||
id: string,
|
||||
): PendingClientRequestRef | undefined {
|
||||
const req = conn.pending.get(id);
|
||||
if (req) return { conn, id, req };
|
||||
return this.registry?.findPendingClientRequest(id);
|
||||
}
|
||||
|
||||
private dropResolvedPermission(conn: AcpConnection, id: string): void {
|
||||
// Delete the exact resolved entry by its `conn.pending` map key. Under
|
||||
// multi-client attach, sibling connections can hold their own pending
|
||||
// entries sharing this one's `bridgeRequestId` (see
|
||||
// ConnectionRegistry.findPendingPermission), so re-matching by
|
||||
// `bridgeRequestId` could delete a sibling's entry and orphan the one we
|
||||
// just resolved. The siblings' now-moot entries are reaped at teardown.
|
||||
conn.pending.delete(id);
|
||||
}
|
||||
|
||||
/**
|
||||
* Drop ONLY the calling connection's own pending permission entry for
|
||||
* `requestId`, never a sibling co-owner's. Under the consensus policy a vote
|
||||
* (or an unexpected vote error) from connection B must not delete connection
|
||||
* A's still-needed entry, which would stall the quorum. A connection that
|
||||
* never streamed the request holds no entry, so this is a no-op for it.
|
||||
*/
|
||||
private dropOwnPendingPermission(
|
||||
conn: AcpConnection,
|
||||
requestId: string,
|
||||
): void {
|
||||
for (const [pid, preq] of conn.pending) {
|
||||
if (preq.kind === 'permission' && preq.bridgeRequestId === requestId) {
|
||||
conn.pending.delete(pid);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle one inbound POST message. Returns nothing — every reply is
|
||||
* delivered asynchronously on a long-lived SSE stream per the RFD
|
||||
|
|
@ -1005,6 +1119,297 @@ export class AcpDispatcher {
|
|||
return;
|
||||
}
|
||||
|
||||
case 'session/permission': {
|
||||
const requestId =
|
||||
typeof params['requestId'] === 'string' ? params['requestId'] : '';
|
||||
if (!requestId) {
|
||||
// Every failure mode below logs to stderr: a stuck permission
|
||||
// prompt is otherwise undebuggable from the server side (the
|
||||
// legacy `resolveClientResponse` path logs its vote/cancel
|
||||
// failures the same way).
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission rejected: \`requestId\` is required`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(id, RPC.INVALID_PARAMS, '`requestId` is required', {
|
||||
httpStatus: 400,
|
||||
requestId,
|
||||
}),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
let response: PermissionResponse;
|
||||
try {
|
||||
response = parsePermissionResponse(params);
|
||||
} catch (err) {
|
||||
// Map the validation throw to a structured 400 here so it carries
|
||||
// the same `{ httpStatus }` envelope as every other error path in
|
||||
// this handler — the outer dispatcher catch would otherwise emit a
|
||||
// plain INVALID_PARAMS with no httpStatus for SDK callers.
|
||||
if (err instanceof AcpParamError) {
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission invalid params (requestId ${logSafe(requestId)}): ${logSafe(err.message)}`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(id, RPC.INVALID_PARAMS, err.message, {
|
||||
httpStatus: 400,
|
||||
requestId,
|
||||
}),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
const sessionIdParam =
|
||||
typeof params['sessionId'] === 'string' &&
|
||||
params['sessionId'].length > 0
|
||||
? params['sessionId']
|
||||
: undefined;
|
||||
// Look up by the globally-unique `requestId` alone, then validate
|
||||
// the client's `sessionId` against the pending entry instead of
|
||||
// trusting it for routing. A mismatched `sessionId` previously fell
|
||||
// through to `sessionIdParam`, which (a) routed `requireOwned` and
|
||||
// the bridge vote at the wrong session and (b) left the real pending
|
||||
// entry to leak until teardown. Reject the mismatch explicitly.
|
||||
const pendingRef = this.registry?.findPendingPermission(requestId);
|
||||
if (
|
||||
sessionIdParam &&
|
||||
pendingRef &&
|
||||
sessionIdParam !== pendingRef.req.sessionId
|
||||
) {
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission vote rejected: requestId ${logSafe(requestId)} does not belong to session ${logSafe(sessionIdParam)}`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(
|
||||
id,
|
||||
RPC.INVALID_PARAMS,
|
||||
'requestId does not belong to the specified session',
|
||||
{ httpStatus: 409, sessionId: sessionIdParam, requestId },
|
||||
),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
// Require a registry hit before voting when the registry is
|
||||
// available. In the scoped `session/permission` route `sessionIdParam`
|
||||
// is always set, so a miss (unknown/stale/already-resolved request)
|
||||
// must NOT fall through to `sessionIdParam` and the bridge — that
|
||||
// reports the bridge's `false` as a thrown 409, whereas
|
||||
// DaemonClient/REST treat a missing request as `404 → false` (not an
|
||||
// exception). Reserve 409 for a present entry the bridge still rejects.
|
||||
if (this.registry && !pendingRef) {
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission vote dropped: no pending request for requestId ${logSafe(requestId)}`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(id, RPC.INVALID_PARAMS, 'No pending permission request', {
|
||||
httpStatus: 404,
|
||||
requestId,
|
||||
}),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
// The pending entry's own session is authoritative; fall back to the
|
||||
// client's `sessionId` only when there is no registry to consult.
|
||||
const sessionId = pendingRef?.req.sessionId ?? sessionIdParam;
|
||||
if (!sessionId) {
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission vote dropped: no pending request for requestId ${logSafe(requestId)}`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(id, RPC.INVALID_PARAMS, 'No pending permission request', {
|
||||
httpStatus: 404,
|
||||
requestId,
|
||||
}),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
// Inline ownership check (not the shared `requireOwned`) so the
|
||||
// rejection carries the same `{ httpStatus }` envelope as every other
|
||||
// error path in this handler — SDK callers classify permission-vote
|
||||
// failures by `error.data.httpStatus`, and this is the likeliest
|
||||
// cross-connection failure (right session header, no `session/new` on
|
||||
// this connection).
|
||||
if (!conn.ownsSession(sessionId)) {
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission vote rejected: session ${logSafe(sessionId)} not owned by this connection (requestId ${logSafe(requestId)})`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(
|
||||
id,
|
||||
RPC.INVALID_PARAMS,
|
||||
'Session not owned by this connection',
|
||||
{ httpStatus: 403, sessionId, requestId },
|
||||
),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
let accepted: boolean;
|
||||
try {
|
||||
accepted = this.bridge.respondToSessionPermission(
|
||||
sessionId,
|
||||
requestId,
|
||||
response,
|
||||
this.sessionCtx(conn, sessionId, loopback),
|
||||
);
|
||||
} catch (err) {
|
||||
// Mirror REST's `sendPermissionVoteError` so ACP SDK callers get the
|
||||
// same shapes for normal permission outcomes instead of a generic
|
||||
// internal error from the outer dispatcher catch: a forged optionId
|
||||
// is a 400 (the requestId is known, the option isn't), a
|
||||
// policy-denied vote is a 403 (well-formed, authenticated, refused).
|
||||
if (err instanceof InvalidPermissionOptionError) {
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission invalid option (${logSafe(sessionId)}, requestId ${logSafe(requestId)}): ${logSafe(err.message)}`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(id, RPC.INVALID_PARAMS, err.message, {
|
||||
httpStatus: 400,
|
||||
code: 'invalid_option_id',
|
||||
requestId: err.requestId,
|
||||
optionId: err.optionId,
|
||||
}),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (err instanceof PermissionForbiddenError) {
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission forbidden (${logSafe(sessionId)}, requestId ${logSafe(requestId)}): ${logSafe(err.message)}`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(id, RPC.INVALID_PARAMS, err.message, {
|
||||
httpStatus: 403,
|
||||
code: 'permission_forbidden',
|
||||
requestId: err.requestId,
|
||||
sessionId: err.sessionId,
|
||||
reason: err.reason,
|
||||
}),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (err instanceof PermissionPolicyNotImplementedError) {
|
||||
// Operator's settings name a policy whose mediator hasn't landed
|
||||
// in this build. 501 (not 500) so the SDK can say "daemon older
|
||||
// than your settings expect; upgrade".
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission policy not implemented (${logSafe(sessionId)}, requestId ${logSafe(requestId)}): ${logSafe(err.message)}`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(id, RPC.INTERNAL_ERROR, err.message, {
|
||||
httpStatus: 501,
|
||||
code: 'permission_policy_not_implemented',
|
||||
policy: err.policy,
|
||||
}),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (err instanceof CancelSentinelCollisionError) {
|
||||
// Agent/daemon contract violation (agent's option set includes
|
||||
// the cancel sentinel), not a client mistake — 500 with a stable
|
||||
// code so the SDK can distinguish it from unrelated internals.
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission cancel-sentinel collision (${logSafe(sessionId)}, requestId ${logSafe(requestId)}): ${logSafe(err.message)}`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(id, RPC.INTERNAL_ERROR, err.message, {
|
||||
httpStatus: 500,
|
||||
code: 'cancel_sentinel_collision',
|
||||
requestId: err.requestId,
|
||||
sentinel: err.sentinel,
|
||||
}),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
// Truly unexpected bridge/sessionCtx failure: the mediator may be
|
||||
// left blocking the agent's prompt. Mirror the legacy
|
||||
// `resolveClientResponse` path — cancel as a fallback (dropping the
|
||||
// entry only if the cancel landed, else keep it for teardown to
|
||||
// retry) — then rethrow so the outer dispatcher catch maps the
|
||||
// error for the wire.
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission vote failed (${logSafe(sessionId)}, requestId ${logSafe(requestId)}): ${logSafe(errMsg(err))}`,
|
||||
);
|
||||
const cancelled = this.cancelAbandonedPermission(
|
||||
{ sessionId, bridgeRequestId: requestId },
|
||||
pendingRef?.conn.sessions.get(sessionId)?.clientId,
|
||||
);
|
||||
// Drop only the VOTING connection's own entry (consistent with the
|
||||
// success path) — never `pendingRef`, which may be a sibling/the
|
||||
// originator. Deleting the originator's entry would stall a quorum
|
||||
// still waiting on its vote. If the voter has no own entry, leave
|
||||
// everything for teardown.
|
||||
if (cancelled) {
|
||||
this.dropOwnPendingPermission(conn, requestId);
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
if (!accepted) {
|
||||
// The bridge mediator had no outstanding request to resolve
|
||||
// (already voted/cancelled — e.g. a duplicate or racing vote).
|
||||
// Do NOT delete the registry entry here: matching the legacy
|
||||
// `resolveClientResponse` contract, keep it until teardown's
|
||||
// `abandonPendingForSession` releases the mediator. Deleting now
|
||||
// would (a) conflate "no pending entry" with "bridge rejected a
|
||||
// present vote" and (b) make a legitimate retry on another
|
||||
// connection fail with a misleading "no pending" error.
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission vote not accepted by bridge (${logSafe(sessionId)}, requestId ${logSafe(requestId)})`,
|
||||
);
|
||||
if (id !== undefined) {
|
||||
conn.sendConn(
|
||||
error(
|
||||
id,
|
||||
RPC.INVALID_PARAMS,
|
||||
'Permission vote not accepted (request already resolved)',
|
||||
{
|
||||
httpStatus: 409,
|
||||
sessionId,
|
||||
requestId,
|
||||
},
|
||||
),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
// Drop ONLY the voting connection's own pending entry for this
|
||||
// request — never the first registry-wide match (`pendingRef`), which
|
||||
// may belong to a sibling connection. Under the consensus policy
|
||||
// `respondToSessionPermission` returns true for an intermediate
|
||||
// "recorded" vote, so deleting a sibling's entry here would drop a
|
||||
// co-owner's still-needed outbound request and could stall the quorum
|
||||
// until timeout/teardown. A cross-connection voter that never streamed
|
||||
// the request has no own entry — leave the originator's for teardown.
|
||||
this.dropOwnPendingPermission(conn, requestId);
|
||||
// Log the success too (every failure branch logs): an operator
|
||||
// grepping a stuck prompt can then tell "vote accepted here" apart
|
||||
// from "vote never arrived" or "vote landed on another connection".
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp session/permission vote accepted (${logSafe(sessionId)}, requestId ${logSafe(requestId)}, connection ${logSafe(conn.connectionId.slice(0, 8))})`,
|
||||
);
|
||||
this.replyConn(conn, id, {});
|
||||
return;
|
||||
}
|
||||
|
||||
// STANDARD method (SDK 0.14.1, non-`unstable_`): model + mode live
|
||||
// here under categories `model`/`mode`, routed to the existing bridge
|
||||
// setters. Replaces the old vendor `_qwen/session/set_model`.
|
||||
|
|
@ -3016,29 +3421,50 @@ export class AcpDispatcher {
|
|||
msg: JsonRpcResponse,
|
||||
fromLoopback: boolean,
|
||||
): void {
|
||||
// Our outbound request ids are strings (`_qwen_perm_N`); a client echoes
|
||||
// Our outbound request ids are strings (`_qwen_perm_<conn>_N`); a client echoes
|
||||
// the same id verbatim. Anything else can't match a pending entry.
|
||||
const id = msg.id;
|
||||
if (typeof id !== 'string') return;
|
||||
const pending = conn.pending.get(id);
|
||||
if (!pending) return;
|
||||
const pendingRef = this.findPendingClientRequest(conn, id);
|
||||
if (!pendingRef) return;
|
||||
const pendingConn = pendingRef.conn;
|
||||
const pending = pendingRef.req;
|
||||
if (pendingConn !== conn && !conn.ownsSession(pending.sessionId)) {
|
||||
// Mirror the `session/permission` handler: never drop a cross-connection
|
||||
// vote silently. The POST already returned 202, so without a log line the
|
||||
// operator has no grep-friendly signal to correlate against a permission
|
||||
// prompt that stays blocked until teardown's `abandonPendingForSession`.
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp permission vote dropped: responding connection ${logSafe(
|
||||
conn.connectionId.slice(0, 8),
|
||||
)} does not own session ${logSafe(pending.sessionId)} (requestId ${logSafe(
|
||||
pending.bridgeRequestId,
|
||||
)})`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
// NOTE: do NOT delete the pending entry yet. Keep it until either the
|
||||
// bridge vote OR the cancel fallback runs — if both somehow fail, the
|
||||
// entry survives so a later session/connection teardown
|
||||
// (`abandonPendingForSession`) can still release the mediator.
|
||||
|
||||
// A client error response is a cancellation; otherwise pass the result
|
||||
// through. The cast defers shape validation to the bridge, so a
|
||||
// MALFORMED result (e.g. `{}` with no `outcome`) makes the mediator
|
||||
// throw — caught below, where we fall back to an explicit cancel so the
|
||||
// mediator is always released. The pending entry is dropped only after a
|
||||
// successful vote/cancel (see the NOTE above), so a double-failure leaves
|
||||
// it for teardown to retry.
|
||||
const vote =
|
||||
'error' in msg
|
||||
? { outcome: { outcome: 'cancelled' } }
|
||||
: (msg as { result: unknown }).result;
|
||||
try {
|
||||
// A client error response is a cancellation; otherwise validate +
|
||||
// whitelist the result through the SAME `parsePermissionResponse` the
|
||||
// `session/permission` handler uses. This PR widened this path to any
|
||||
// co-owning connection (via `findPendingClientRequest`), so without the
|
||||
// shared validator a co-owner could inject arbitrary top-level args and
|
||||
// extra `outcome` sub-fields straight to the bridge. A malformed result
|
||||
// throws here (as the old direct cast made the bridge throw) and is
|
||||
// caught below, where the cancel fallback always releases the mediator.
|
||||
const vote =
|
||||
'error' in msg
|
||||
? { outcome: { outcome: 'cancelled' } }
|
||||
: parsePermissionResponse(
|
||||
isObject((msg as { result: unknown }).result)
|
||||
? (msg as { result: Record<string, unknown> }).result
|
||||
: {},
|
||||
);
|
||||
this.bridge.respondToSessionPermission(
|
||||
pending.sessionId,
|
||||
pending.bridgeRequestId,
|
||||
|
|
@ -3047,7 +3473,7 @@ export class AcpDispatcher {
|
|||
>[2],
|
||||
this.sessionCtx(conn, pending.sessionId, fromLoopback),
|
||||
);
|
||||
conn.pending.delete(id); // vote landed — safe to drop
|
||||
this.dropResolvedPermission(pendingConn, id);
|
||||
} catch (err) {
|
||||
writeStderrLine(
|
||||
`qwen serve: /acp permission vote failed (${logSafe(pending.sessionId)}): ${logSafe(errMsg(err))}`,
|
||||
|
|
@ -3058,9 +3484,9 @@ export class AcpDispatcher {
|
|||
// permanently stuck with no recovery path.
|
||||
const cancelled = this.cancelAbandonedPermission(
|
||||
pending,
|
||||
conn.sessions.get(pending.sessionId)?.clientId,
|
||||
pendingConn.sessions.get(pending.sessionId)?.clientId,
|
||||
);
|
||||
if (cancelled) conn.pending.delete(id);
|
||||
if (cancelled) this.dropResolvedPermission(pendingConn, id);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -280,18 +280,24 @@ export function mountAcpHttp(
|
|||
if (!enabled) return undefined;
|
||||
|
||||
const path = opts.path ?? '/acp';
|
||||
const dispatcher = new AcpDispatcher(
|
||||
bridge,
|
||||
opts.boundWorkspace,
|
||||
opts.workspace,
|
||||
opts.fsFactory,
|
||||
opts.deviceFlowRegistry,
|
||||
opts.sessionShellCommandEnabled === true,
|
||||
);
|
||||
const dispatcherRef: { current?: AcpDispatcher } = {};
|
||||
// When a session/connection tears down with a permission still pending,
|
||||
// cancel it on the bridge so the agent's prompt isn't left blocked.
|
||||
const registry = new ConnectionRegistry(
|
||||
(req, clientId) => dispatcher.cancelAbandonedPermission(req, clientId),
|
||||
(req, clientId) => {
|
||||
// Defensive, matching the `detachClient` callback below: if a future
|
||||
// refactor introduces async work between registry and dispatcher
|
||||
// creation, a teardown racing in here must not crash
|
||||
// `abandonPendingForSession`. Log and report "not cancelled" instead of
|
||||
// throwing through the teardown path.
|
||||
if (!dispatcherRef.current) {
|
||||
writeStderrLine(
|
||||
'qwen serve: /acp abandonPending called before dispatcher initialized (skipped)',
|
||||
);
|
||||
return false;
|
||||
}
|
||||
return dispatcherRef.current.cancelAbandonedPermission(req, clientId);
|
||||
},
|
||||
// Best-effort bridge detach so a torn-down connection's bridge-stamped
|
||||
// client ids don't linger in the bridge's voter/known-client sets.
|
||||
(sessionId, clientId) => {
|
||||
|
|
@ -305,6 +311,16 @@ export function mountAcpHttp(
|
|||
},
|
||||
opts.maxConnections,
|
||||
);
|
||||
const dispatcher = new AcpDispatcher(
|
||||
bridge,
|
||||
opts.boundWorkspace,
|
||||
opts.workspace,
|
||||
opts.fsFactory,
|
||||
opts.deviceFlowRegistry,
|
||||
opts.sessionShellCommandEnabled === true,
|
||||
registry,
|
||||
);
|
||||
dispatcherRef.current = dispatcher;
|
||||
|
||||
// ── POST /acp ──────────────────────────────────────────────────────
|
||||
app.post(path, async (req: Request, res: Response) => {
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
Loading…
Add table
Add a link
Reference in a new issue