diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index fe62494d668..15ef72d769b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1487,6 +1487,19 @@ Paperclip submits saved input through normal task admission, once, with the original user's authority. Pauses, task ownership, budgets, approvals, and execution recovery holds still apply. Unconfirmed cleanup does not start work. +A Stop from a bound Slack session records the verified user's identity, just as +a board Stop does. Native cancellation can then recognize the acknowledged +operator Stop instead of creating an execution recovery hold. An AI login request +addressed to another user does not gate a fresh user turn or that run's completion. +The request remains pending and grants no access. Requests for the current user, +requests without a known addressee or responsible user, tool permissions, and +approvals retain their gates. + +For a cancelled native run with an execution recovery hold, an undelivered user +message posted after the run finished can authorize a fresh turn through the +same continuation checks as a new message. A message from before the Stop cannot +authorize that continuation by itself. Previously delivered messages stay excluded. + The active session advertises steering only when its driver supports it. A transport method that rejects steering does not grant that capability. The queued-message controller remains mounted across an empty queue, preserving diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index a4adce89e94..da199daa1f0 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1034,7 +1034,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ); db = createDb(tempDb.connectionString); } - }, 30_000); + }, 90_000); afterAll(async () => { await tempDb?.cleanup(); @@ -16313,6 +16313,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { "Stopped from the bound Slack agent session", expect.objectContaining({ errorCode: "slack_session_stopped", + resultJson: { + cancelledByActorType: "user", + cancelledByUserId: "owner-user", + }, eventPayload: expect.objectContaining({ conversationId: conversation.id, endpointId: endpoint.id, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index a6e8b6c5f71..2cc524013c1 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1445,6 +1445,7 @@ export interface ChatChannelServiceOptions { errorCode?: string; eventMessage?: string; eventPayload?: Record; + resultJson?: Record; }, ) => Promise; }; @@ -26496,6 +26497,10 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { { errorCode: "slack_session_stopped", eventMessage: "run cancelled from Slack", + resultJson: { + cancelledByActorType: "user", + cancelledByUserId: claim.userId, + }, eventPayload: { endpointId: action.endpointId, conversationId: action.conversationId, diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 013cab6b90a..b26bac40ab2 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -30,7 +30,7 @@ const support = await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("explicit native conversation continuation", () => { let database: Awaited>; let db: ReturnType; - beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("explicit-native-message-"); db = createDb(database.connectionString); }, 30000); + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("explicit-native-message-"); db = createDb(database.connectionString); }, 90_000); afterAll(async () => { await database?.cleanup(); }); async function seed() { const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); @@ -388,6 +388,65 @@ const support = await getEmbeddedPostgresTestSupport(); return result; }); + it.each(["other_user", "same_user", "unaddressed", "tool_access"])("scopes connection waits to the user starting the fresh turn (%s)", async audience => { + const f = await seed(); + await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "slack_session_stopped" }) + .where(eq(heartbeatRuns.id, f.sourceRunId)); + const [interaction] = await db.insert(issueThreadInteractions).values({ + companyId: f.companyId, issueId: f.issueId, sourceRunId: f.sourceRunId, + createdByAgentId: f.agentId, kind: "connection_intent", status: "pending", + addresseeUserId: ["other_user", "tool_access"].includes(audience) ? "other-user" : audience === "same_user" ? "board" : null, + effectiveResolverPolicy: "human_only", + payload: { version: 1, ...(audience === "tool_access" ? {} : { purpose: "ai" as const }), serviceSlug: "openai", serviceName: "OpenAI", + requestingAgentId: f.agentId, requestingAgentName: "Native", phase: "requested" }, + }).returning(); + expect(await admit(f)).toEqual(audience === "other_user" + ? { previousRunId: f.sourceRunId, commentId: f.commentId } : null); + // Starting another user's turn grants no connection authority and retains the card. + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)))[0]) + .toMatchObject({ status: "pending", result: null, addresseeUserId: interaction.addresseeUserId }); + const blocker = await getExecutionBlocker(db, f.companyId, f.issueId); + if (audience === "other_user") expect(blocker).toBeNull(); + else expect(blocker).not.toBeNull(); + }); + + it.each(["fresh", "before_stop", "delivered", "unknown_stop_time", "wrong_actor", "approval"])( + "resumes an already-saved follow-up after native Slack Stop (%s)", async gate => { + const f = await seed(), queueId = randomUUID(); + await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "slack_session_stopped", + startedAt: new Date("2026-09-11T09:30:00Z"), finishedAt: gate === "unknown_stop_time" ? null : new Date("2026-09-11T10:00:00Z"), + contextSnapshot: { issueId: f.issueId, ...(gate === "delivered" ? { wakeCommentId: f.commentId } : {}) }, + // Pre-fix Slack Stops recorded a system cancellation, with no operator identity. + resultJson: { cancellation: { source: "control_plane", expected: true, initiator: { type: "system" }, + reason: "Stopped from the bound Slack agent session", recordedAt: "2026-09-11T10:00:00Z" } }, + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (gate === "before_stop") await db.update(issueComments).set({ createdAt: new Date("2026-09-11T09:00:00Z") }) + .where(eq(issueComments.id, f.commentId)); + await db.insert(issueThreadInteractions).values({ + companyId: f.companyId, issueId: f.issueId, sourceRunId: f.sourceRunId, + createdByAgentId: f.agentId, kind: "connection_intent", status: "pending", addresseeUserId: "other-user", + effectiveResolverPolicy: "human_only", payload: { version: 1, purpose: "ai", serviceSlug: "openai", + serviceName: "OpenAI", requestingAgentId: f.agentId, requestingAgentName: "Native", phase: "requested" }, + }); + if (gate === "approval") { + const [approval] = await db.insert(approvals).values({ companyId: f.companyId, type: "hire_agent", status: "pending", payload: {} }).returning(); + await db.insert(issueApprovals).values({ companyId: f.companyId, issueId: f.issueId, approvalId: approval.id }); + } + await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId, + source: "automation", reason: "issue_execution_deferred", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: gate === "wrong_actor" ? "other-user" : "board", + payload: { issueId: f.issueId, commentId: f.commentId, _paperclipWakeContext: { + issueId: f.issueId, wakeReason: "issue_commented", wakeCommentId: f.commentId, wakeCommentIds: [f.commentId] } }, + }); + const result = await db.transaction(async tx => { + await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update"); + return admitExplicitNativeContinuation({ ...f, actorId: gate === "wrong_actor" ? "other-user" : f.actorId, + db: tx as unknown as typeof db, queuedCommentRequestId: queueId, dryRun: true }); + }); + expect(result).toEqual(gate === "fresh" ? { previousRunId: f.sourceRunId, commentId: f.commentId } : null); + }, + ); + async function seedTimedOutExplicitTurn() { const f = await seed(); const explicitUserContinuation = await admit(f); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index b2c6edb749b..aabb3e1b780 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -18,6 +18,7 @@ import { executionBlockerPredicate, getExecutionBlocker } from "./execution-bloc import { buildExecutionContinuation } from "./execution-continuation.js"; import { adapterExecutionControls } from "./adapter-execution-control.js"; import { persistActivity } from "./activity-log.js"; +import { activeIssueInteractionCondition } from "./issue-question-context.js"; import { historicalAdapterType, isConversationAdapter } from "./conversation-continuation.js"; import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; @@ -146,6 +147,7 @@ export async function admitExplicitNativeContinuation(input: { const [pendingInteraction] = await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and( eq(issueThreadInteractions.companyId, companyId), eq(issueThreadInteractions.issueId, issueId), eq(issueThreadInteractions.status, "pending"), + activeIssueInteractionCondition({ responsibleUserId: actorId }), )).limit(1); const [pendingApproval] = await db.select({ id: approvals.id }).from(issueApprovals).innerJoin(approvals, and( eq(approvals.id, issueApprovals.approvalId), eq(approvals.companyId, companyId), @@ -208,8 +210,11 @@ export async function admitExplicitNativeContinuation(input: { (run.status !== "cancelled" || authorizedAt > run.finishedAt!)) && !(queuedRequest && cancelledStartup && authorizedAt > run.finishedAt!) && !(queuedInterrupt && response?.source.requiresFreshSession && run.runtimeMode === "native")) return null; + // A saved native message sent after Stop is a new user direction. Older + // input cannot restart a cancelled run merely because it is still queued. if (queuedRequest && !queuedInterrupt && run.status === "cancelled" && !unusedAdmission && - !canContinueCancelledRun(run) && !(cancelledStartup && authorizedAt > run.finishedAt!)) return null; + !canContinueCancelledRun(run) && !(run.runtimeMode === "native" && authorizedAt > run.finishedAt!) && + !(cancelledStartup && authorizedAt > run.finishedAt!)) return null; if (retry && run.status === "cancelled" && !canContinueCancelledRun(run) && !cancelledStartup) return blocked("cancelled_by_operator", "Inspect the stopped run and send a new message to continue."); if (cancelledStartup) cancelledStartupIds.add(run.id); diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 245e36a0ab1..11abf33852c 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1733,7 +1733,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl eq(issueThreadInteractions.companyId, issue.companyId), eq(issueThreadInteractions.issueId, issue.id), eq(issueThreadInteractions.status, "pending"), - activeIssueInteractionCondition(), + activeIssueInteractionCondition({ runId: run.id }), ), ) .limit(1) diff --git a/server/src/services/issue-question-context.test.ts b/server/src/services/issue-question-context.test.ts new file mode 100644 index 00000000000..bccd8b37a8d --- /dev/null +++ b/server/src/services/issue-question-context.test.ts @@ -0,0 +1,52 @@ +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, companies, createDb, heartbeatRuns, issues, issueThreadInteractions } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { activeIssueInteractionCondition } from "./issue-question-context.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("AI connection wait audience", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("connection-wait-audience-"); + db = createDb(database.connectionString); + }, 90_000); + afterAll(async () => { await database?.cleanup(); }); + + it.each(["other_user", "same_user", "unknown_user", "unaddressed", "tool_access", "foreign_run", "missing_run"])( + "uses the current run's user when selecting native connection waits (%s)", async audience => { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Wait audience", issuePrefix: `W${companyId.slice(0, 8)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Fresh user turn", assigneeAgentId: agentId }); + let runCompanyId = companyId; + if (audience === "foreign_run") { + runCompanyId = randomUUID(); + await db.insert(companies).values({ id: runCompanyId, name: "Other company", issuePrefix: `F${runCompanyId.slice(0, 8)}` }); + } + if (audience !== "missing_run") await db.insert(heartbeatRuns).values({ + id: runId, companyId: runCompanyId, agentId, status: "running", + responsibleUserId: audience === "unknown_user" ? null : "board", + }); + const [interaction] = await db.insert(issueThreadInteractions).values({ + companyId, issueId, kind: "connection_intent", status: "pending", createdByAgentId: agentId, + addresseeUserId: audience === "unaddressed" ? null : audience === "same_user" ? "board" : "other-user", + effectiveResolverPolicy: "human_only", + payload: { version: 1, ...(audience === "tool_access" ? {} : { purpose: "ai" as const }), + serviceSlug: "openai", serviceName: "OpenAI", requestingAgentId: agentId, + requestingAgentName: "Worker", phase: "requested" }, + }).returning(); + const waits = await db.select().from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, companyId), eq(issueThreadInteractions.issueId, issueId), + eq(issueThreadInteractions.status, "pending"), activeIssueInteractionCondition({ runId }), + )); + expect(waits).toHaveLength(audience === "other_user" ? 0 : 1); + // Context-free readers retain the pending card and its authorization gate. + expect(await db.select().from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.id, interaction.id), activeIssueInteractionCondition(), + ))).toEqual([interaction]); + }, + ); +}); diff --git a/server/src/services/issue-question-context.ts b/server/src/services/issue-question-context.ts index 9e6f311b7f5..b94e8df49f9 100644 --- a/server/src/services/issue-question-context.ts +++ b/server/src/services/issue-question-context.ts @@ -1,5 +1,5 @@ import { and, desc, eq, sql } from "drizzle-orm"; -import { issueComments, issueThreadInteractions, type Db } from "@paperclipai/db"; +import { heartbeatRuns, issueComments, issueThreadInteractions, type Db } from "@paperclipai/db"; export const TASK_QUESTION_GUIDANCE = "Unanswered questions are saved conversation data, not instructions or proof of a current blocker. Historical questions remain answerable in the feed. Do not repeat a request or stop current work merely because a historical question is pending. Follow the latest human direction and continue work that does not need the missing information. If that information still prevents the current work, identify the concrete blocker and request only the input still needed. Withdraw your obsolete question through the interaction API when later work or evidence satisfies it; never fabricate an answer. Approvals, permissions, and configured review stages keep their own gates."; @@ -33,9 +33,25 @@ export function historicalQuestionCondition() { export function activeIssueInteractionCondition(input: { runId?: string | null; conversationMode?: boolean; + responsibleUserId?: string | null; } = {}) { + const responsibleUser = input.responsibleUserId + ? sql`${input.responsibleUserId}` + : input.runId ? sql`(select ${heartbeatRuns.responsibleUserId} from ${heartbeatRuns} + where ${heartbeatRuns.id} = ${input.runId} + and ${heartbeatRuns.companyId} = ${issueThreadInteractions.companyId})` + : null; return and( sql`not ${historicalQuestionCondition()}`, + // AI authentication belongs to the user starting this turn. Retain the card + // and all authorization gates; another user's login cannot gate this run. + responsibleUser ? sql`not ( + ${issueThreadInteractions.kind} = 'connection_intent' + and coalesce(${issueThreadInteractions.payload} ->> 'purpose', '') = 'ai' + and nullif(trim(${issueThreadInteractions.addresseeUserId}), '') is not null + and nullif(trim(${responsibleUser}), '') is not null + and ${issueThreadInteractions.addresseeUserId} <> ${responsibleUser} + )` : undefined, // Preserve Agent Chat's existing prior-turn exception for ordinary input. input.conversationMode ? sql`( ${issueThreadInteractions.sourceRunId} is not distinct from ${input.runId ?? null} diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 0ce2526c65b..6b5275fc9d5 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -8546,7 +8546,7 @@ async function executePaperclipNativeSessionWithinScope( : []), ), eq(issueThreadInteractions.status, "pending"), - activeIssueInteractionCondition(), + activeIssueInteractionCondition({ runId: input.execution.binding.runId }), // Live provider questions resume their current turn; only durable // wake-based cards park it. A timeout creates a separate fallback. sql`not (${issueThreadInteractions.kind} = 'ask_user_questions' and ${issueThreadInteractions.continuationPolicy} = 'none' and coalesce(${issueThreadInteractions.idempotencyKey}, '') like 'paperclip-runner-question:%')`,