Files
PaperClipAI/server/src/services/native-runtime/native-interaction-bridge.ts
T
ec3b558277 fix(server): schedule accepted native response recovery (#17696)
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Native runs store accepted results and their finalization decisions.
> - A completed run can still owe its public answer after an
interruption.
> - Periodic recovery scans for that answer even when no new work
exists.
> - This pull request wakes recovery from committed changes and saved
retry deadlines.
> - The existing owner still verifies the result, source, and current
publication authority.
> - Settled or dormant response work no longer needs its own periodic
scan.

## Linked Issues or Issue Description

Refs #17690 (merged). This is slice 48 of the sleep migration stack, now
based on master. Related: #17661 handles GitHub terminal publication;
this change handles native accepted-response presentation.
Governed-review presentation remains a separate duty.

**What happened?**

Native finalization recovery repeatedly checks already accepted results
for a missing answer. Removing that check without complete notifications
would lose responses when evidence arrives late, an agent resumes, or
admission settles after execution.

**Expected behavior**

Recover the answer from its saved result when a supported change makes
it eligible. Retain real retry deadlines. Stop querying when the answer
is selected or cannot currently be published. Reconstruct work after
restart.

**Steps to reproduce**

1. Commit a native accepted result with no selected response comment.
2. Run finalization recovery while its source is unavailable or locked.
3. Restore that source or release the lock. The answer must appear once
without repeating provider execution.

## What Changed

- Honor recorded non-create decisions and skip-comment policy in both
discovery and recovery. Apply durable conversation/result/continuation
guards before recovery locks. Preserve genuine unanswered and legacy
recovery.
- Scope routine response wakes to the affected issue/run. Coalesced
input names its active run; pending-effect settlement names its issue.
Sixteen existing global authority/configuration hooks use one explicit
company invalidation boundary. No new transaction wrappers or tables are
added.
- Add an accepted-response worker using the existing delivery
coordinator and scoped deadline helper. Preserve saved retry deadlines
and pending-effect fences under the owner's lock.
- Share the finalization notification boundary. Wake on finalization
settlement, exact accepted-result/final-message evidence, agent
availability, retry admission, and existing chat authority changes.
- Keep lifecycle policy in services. A narrow pre-write status hook lets
lifecycle retain ownership of its transaction and status transitions.
- Register the worker during enabled maintenance startup. Remove its
authority subscription on shutdown. Periodic replay skips this duty on
both committed and fresh-commit paths; ordinary finalization and
explicit recovery retain their behavior.
- Add a partial index and expression statistics for unpresented native
runs. Migration 0412 supports concurrent replay and adds no queue table.
- Add real PostgreSQL and lifecycle coverage. Preserve the
implementation and producer audit in the plan document.

## Verification

- Additional real Photon callback acceptance on the published production
code disproves the security warning: ten active assertions cause zero
response-recovery hints/SELECTs, while three concurrent assertions
during an actual attention-to-active recovery cause exactly one
wake/SELECT. The existing locked status guard already enforces this
boundary. No production change was needed; the resolved finding now has
a successful current-head security check. The regression will be
retained with the next slice.
- Eligibility repair source `892f36e5f66ba1af123034936a7160c64c3ab883`
passes **1489 distinct server cases**: 1139 in six complete
recovery/finalizer/monitor/summary/workspace suites (81.02 seconds),
plus 350 selected chat cases (297.68s). Another 1463 chat cases are
outside the pattern; the complete 1813-case file is not claimed. All
eleven new regressions are included in that count. Five dedicated fresh
PostgreSQL 17 databases migrate through unchanged 0412, with no
fixture-source adaptations. Database build, server types/build,
boundaries and full-SHA migration ordering pass. All seven test source
hashes match.
- The exact published baseline first fails eight behavioral assertions
while the three genuine-answer cases pass. The same corrected test
source passes all eleven after repair: recorded none/reuse remain
suppressed; five ineligible history families take zero recovery row
locks across repeated issue hints; missing/legacy/create-null answers
still publish once; fresh periodic finalization defers presentation to
its worker. Earlier fixture setup and variable/compiler failures remain
explained in the audit, rather than erased by later success.
- Repair source `2fc8b0614aaf9ac7337045de4ced7325f95edc45` passes **1821
server cases**: 1496 in 13 complete producer/lifecycle/authority suites
(110.63 seconds), plus 325 selected chat cases (278.99 seconds).
Coverage includes all native-session cases, queue admission, status
decisions, membership/board access, the complete GitHub workflow,
accepted-response lifecycle and real authority restoration. The other
1477 chat cases are outside this selection; the complete 1802-case file
is not claimed for this repair run.
- Actual PostgreSQL regressions first reproduce a coalesced-input hint
missing its run scope and two routine hint paths producing 12
response-history SELECTs each after three notifications. The same
assertions pass with run-scoped coalescing and zero response-history
SELECTs for unscoped routine hints. Actual membership restoration still
publishes the waiting response once. Assertion time limits are
unchanged; this measures real driver SQL and persisted producer work.
- Eleven dedicated fresh PostgreSQL 17 databases migrate through 0412.
One embedded-only wake-queue fixture import temporarily uses its own
external database and restores byte for byte. All 14 selected source
files match pre-run hashes. This qualifies external-database behavior
rather than reproducing embedded CI locally.
- Master advanced independently with #18126's Pi archive retry fix.
Normal merge `2a535bc61259ef99387369e8b32498f774e5ba2c` preserves both
complete binary patches. All accepted server/database/UI/test/CLI trees
stay identical; the three independent runner installer/manifest files
are retained. All 13 installer tests pass, including stalled header/body
cancellation. Combined server types/build, boundaries and full-SHA
migration order pass. A subsequent normal merge of #18128 at
`e8b78cce5bb557b086bee9ca930eeaab1d09b244` retains its three test-only
file changes with both complete patches preserved. The changed HTTPS
backfill case passes on a fresh PostgreSQL fixture (1 selected case; 178
outside the pattern), and combined types/build, boundaries and migration
order pass again. All 14 repair source hashes remain unchanged. The
independent browser fixture changes are preserved; no local browser run
is claimed.
- The earlier source `9feec170ef8174011fc3c5d3d6d76f0a0f39fea0` passed
the 4171-case full baseline before this scope repair. That scope is
preserved in the audit rather than represented as a current full rerun.
All 0412 database bytes remain unchanged: original concurrent
index/statistics/ANALYZE, prior executable history, journal entries and
retained snapshots. Five snapshots 0408 through 0412 remain. The
populated 0411-to0412 upgrade preserved 410 prior receipts and 38
companies, appended one receipt and none on replay; index/statistic
checks passed.
- The earlier P2 security failure, Greptile's former conditional
dependency on #17702, and the subsequent actual Ponytail eligibility
findings are recorded in
`doc/plans/2026-10-25-native-response-presentation-sleep-audit.md`. This
PR now includes the issue-scoped pending-effect notification fix itself,
so it is independently correct before #17702. Fresh published-head CI,
Greptile 5/5 and a clean security scan remain required. No
deployed-sleep, external-wake or fleet-savings claim is made.


## Risks

- The writer contract is process-local. Independent SQL writers and
other replicas are not covered by these notifications.
- Unpresented backlog reconstruction still has memory and time cost
proportional to candidate count. Pages contain at most 100 rows;
execution uses the shared fair batch. Selected history is excluded.
Health-only signals do not trigger reconstruction.
- Publication policy remains with its current owner. Digest, company/run
binding, source authorization, selected-answer suppression, and
ownership fences remain required.
- The rest of native recovery and governed-review presentation still run
separately. This PR does not relax Cloud sleep eligibility or add
external wake coverage.
- The stack parent has merged and its actual ancestry is reconciled.
This publication still needs fresh current-head CI and review.

## Model Used

OpenAI Codex, GPT-6 family, with reasoning, repository tools, shell
execution, and PostgreSQL tests. The runtime does not expose a more
specific model identifier or context-window size.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
Co-authored-by: Codex <noreply@openai.com>
2026-10-29 02:31:20 -05:00

968 lines
41 KiB
TypeScript

import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
import type { Db } from "@paperclipai/db";
import type { NativeInteractionResponseEnvelope } from "../../vendor/paperclip-runner/index.js";
import { and, asc, eq, inArray } from "drizzle-orm";
import {
agents,
toolActionRequests,
toolInvocations,
heartbeatRuns,
issues,
issueThreadInteractions,
nativeRunFinalizations,
nativeRunResults,
statusDecisions,
workAssessments,
} from "@paperclipai/db";
import { getAgentWorkEligibility } from "@paperclipai/shared";
import { issueThreadInteractionService } from "../issue-thread-interactions.js";
import { commitNativeStatusDecision } from "./status-decision-committer.js";
import { nativeSha256 } from "./canonical.js";
import { recordNativeAttentionAssessment } from "./work-assessments.js";
import { formatDurableQuestionResponseSummary } from "../question-response-delivery.js";
import { findSatisfiedToolConnection } from "../satisfied-connection-intents.js";
import {
NATIVE_STATUS_ARBITER_POLICY_VERSION,
type NativeAuthoritativeIssueStatus,
type NativeGovernanceGate,
type NativeStatusDecision,
type NativeStatusEffect,
} from "./status-arbiter.js";
export class NativeInteractionBridgeError extends Error {
constructor(readonly code: string, message: string) {
super(message);
this.name = "NativeInteractionBridgeError";
}
}
function record(value: unknown): Record<string, unknown> {
return value && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown>
: {};
}
export type NativeAttentionMaterializedTarget = {
effectKind: string;
targetType: string;
targetId: string;
};
/**
* Applies audit-only attention outcomes that intentionally create no status
* decision. The durable interaction/coordinator mutation is the proof; a
* synthetic delivered effect row is neither created nor accepted.
*/
export async function applyNativeAttentionStatusDecision(input: {
db: Db;
companyId: string;
issueId: string;
runId: string;
decision: NativeStatusDecision;
targetInteractionId?: string;
canonicalRequestId?: string;
}): Promise<NativeAttentionMaterializedTarget[]> {
if (input.decision.statusAction !== "preserve") {
throw new NativeInteractionBridgeError("native_attention_status_authority_invalid", "Audit-only attention cannot change issue status");
}
return input.db.transaction(async (tx) => {
const targets: NativeAttentionMaterializedTarget[] = [];
for (const effect of input.decision.effects) {
if (effect.kind === "link_canonical_request") {
const boundIds = [input.canonicalRequestId, input.targetInteractionId]
.filter((id): id is string => typeof id === "string");
const interactions = await tx.select({
id: issueThreadInteractions.id,
summary: issueThreadInteractions.summary,
})
.from(issueThreadInteractions).where(and(
eq(issueThreadInteractions.companyId, input.companyId),
eq(issueThreadInteractions.issueId, input.issueId),
...(boundIds.length > 0 ? [inArray(issueThreadInteractions.id, boundIds)] : []),
)).limit(2);
const canonical = input.canonicalRequestId
? interactions.find((interaction) => interaction.id === input.canonicalRequestId)
: interactions[0];
const duplicate = input.targetInteractionId
? interactions.find((interaction) => interaction.id === input.targetInteractionId)
: interactions.find((interaction) => interaction.id !== canonical?.id);
if (!canonical || !duplicate || canonical.id === duplicate.id) {
throw new NativeInteractionBridgeError("native_attention_canonical_pair_missing", "Canonical and duplicate requests are required");
}
const summary = `Canonical native attention request: ${canonical.id}`;
if (duplicate.summary !== summary) {
await tx.update(issueThreadInteractions).set({
summary,
updatedAt: new Date(),
}).where(eq(issueThreadInteractions.id, duplicate.id));
}
targets.push({ effectKind: effect.kind, targetType: "issue_thread_interaction", targetId: duplicate.id });
continue;
}
if (effect.kind === "record_stale_response") {
const interaction = await tx.select({
id: issueThreadInteractions.id,
summary: issueThreadInteractions.summary,
})
.from(issueThreadInteractions).where(and(
eq(issueThreadInteractions.companyId, input.companyId),
eq(issueThreadInteractions.issueId, input.issueId),
...(input.targetInteractionId ? [eq(issueThreadInteractions.id, input.targetInteractionId)] : []),
)).limit(1).then((rows) => rows[0] ?? null);
if (!interaction) throw new NativeInteractionBridgeError("native_stale_response_missing", "Stale response is not durable");
const summary = "Response retained for audit after native supersession.";
if (interaction.summary !== summary) {
await tx.update(issueThreadInteractions).set({
summary,
updatedAt: new Date(),
}).where(eq(issueThreadInteractions.id, interaction.id));
}
targets.push({ effectKind: effect.kind, targetType: "issue_thread_interaction", targetId: interaction.id });
continue;
}
if (effect.kind === "record_finalization_error") {
if (effect.cause === "native_workspace_sync_out_unsafe_archive") {
await notifyUnsafeWorkspaceExportRecovery(tx, input);
}
const [coordinator] = await tx.update(nativeRunFinalizations).set({
phase: "terminal_failure",
failureCode: effect.cause,
failureDetail: { nextAction: effect.nextAction, recoveryOwner: { kind: "agent", agentId: effect.agentId } },
nextAttemptAt: null,
updatedAt: new Date(),
}).where(and(
eq(nativeRunFinalizations.runId, input.runId),
eq(nativeRunFinalizations.companyId, input.companyId),
eq(nativeRunFinalizations.issueId, input.issueId),
)).returning({ runId: nativeRunFinalizations.runId });
if (!coordinator) throw new NativeInteractionBridgeError("native_finalization_coordinator_missing", "Native coordinator is required");
await tx.update(heartbeatRuns).set({
nativePhase: "terminal_failure",
nativePhaseUpdatedAt: new Date(),
errorCode: effect.cause,
updatedAt: new Date(),
}).where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId)));
targets.push({ effectKind: effect.kind, targetType: "native_run_finalization", targetId: coordinator.runId });
continue;
}
throw new NativeInteractionBridgeError("native_attention_effect_unimplemented", effect.kind);
}
return targets;
});
}
/**
* Materialize only interactions that authorized this run's wake. Resolution
* remains owned by the existing interaction service; this bridge is a
* read-only, identity-bound projection and never receives credentials.
*/
export async function materializeNativeInteractionResponses(input: {
db: Db;
companyId: string;
issueId: string;
runId: string;
agentId: string;
interactionIds: string[];
}): Promise<NativeInteractionResponseEnvelope[]> {
const requestedIds = new Set(input.interactionIds.filter(Boolean));
if (requestedIds.size === 0) return [];
const [interactions, issue] = await Promise.all([
issueThreadInteractionService(input.db).listForIssue(input.issueId),
input.db.select().from(issues).where(and(
eq(issues.id, input.issueId),
eq(issues.companyId, input.companyId),
)).limit(1).then((rows) => rows[0] ?? null),
]);
if (!issue) throw new NativeInteractionBridgeError("native_interaction_binding_mismatch", "Issue binding not found");
const responses: NativeInteractionResponseEnvelope[] = [];
for (const interaction of interactions) {
if (!requestedIds.has(interaction.id)) continue;
if (interaction.companyId !== input.companyId || interaction.issueId !== input.issueId) {
throw new NativeInteractionBridgeError(
"native_interaction_binding_mismatch",
`Interaction ${interaction.id} is not bound to the native company and issue`,
);
}
if (interaction.kind === "request_confirmation" && interaction.payload.toolAction) {
const action = interaction.payload.toolAction;
if (["accepted", "rejected"].includes(interaction.status) && (interaction.resolvedByAgentId || interaction.resolvedByRunId === input.runId)) {
throw new NativeInteractionBridgeError("native_interaction_self_approval", "Agents cannot resolve governed tool reviews");
}
const [request] = await input.db.select().from(toolActionRequests).where(and(eq(toolActionRequests.id, action.actionRequestId), eq(toolActionRequests.companyId, input.companyId), eq(toolActionRequests.issueId, input.issueId), eq(toolActionRequests.interactionId, interaction.id), eq(toolActionRequests.invocationId, action.invocationId)));
const [invocation] = await input.db.select().from(toolInvocations).where(and(eq(toolInvocations.id, action.invocationId), eq(toolInvocations.companyId, input.companyId), eq(toolInvocations.issueId, input.issueId), eq(toolInvocations.agentId, input.agentId)));
if (!request || !invocation || request.requestedByAgentId !== input.agentId || request.canonicalArgumentsHash !== action.argumentsHash) {
throw new NativeInteractionBridgeError("native_interaction_governed_request_unresolved", "Tool review has no matching authoritative invocation");
}
if (["expired", "cancelled"].includes(request.status)) {
if (interaction.status !== request.status && !(interaction.status === "accepted" && interaction.result?.toolAction?.status === "expired")) throw new NativeInteractionBridgeError("native_interaction_governed_result_mismatch", "Tool review lifecycle does not match its request");
responses.push({ interactionId: interaction.id, kind: interaction.kind, response: { status: interaction.status, result: structuredClone(interaction.result), executionStatus: request.status } });
continue;
}
if (!request.decidedByUserId || request.decidedByUserId !== interaction.resolvedByUserId || !["executed", "failed", "rejected"].includes(request.status) || (request.status === "rejected" ? interaction.status !== "rejected" : interaction.status !== "accepted")) {
throw new NativeInteractionBridgeError("native_interaction_governed_request_unresolved", "Tool review must have a human decision and an authoritative terminal execution outcome");
}
const expectedInvocationStatus = request.status === "executed" ? "succeeded" : request.status === "rejected" ? "denied" : "failed";
if (invocation.status !== expectedInvocationStatus || (request.status !== "rejected" && interaction.result?.toolAction?.status !== request.status)) throw new NativeInteractionBridgeError("native_interaction_governed_result_mismatch", "Tool review outcome does not match its invocation");
}
const interactionResult = record(interaction.result);
if (interaction.kind === "connection_intent" && interaction.status === "expired"
&& interactionResult.outcome === "expired" && typeof interactionResult.connectionId === "string") {
const [run] = await input.db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns)
.where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.agentId, input.agentId)));
const ready = issue.assigneeAgentId === input.agentId && run?.responsibleUserId === interaction.addresseeUserId
? await findSatisfiedToolConnection(input.db, issue, { ...interaction, addresseeUserId: interaction.addresseeUserId ?? null, status: "pending" }, input.runId) : null;
if (!ready || ready.id !== interactionResult.connectionId) {
throw new NativeInteractionBridgeError("native_interaction_unresolved", "The existing connection is no longer available");
}
responses.push({ interactionId: interaction.id, kind: interaction.kind,
response: { status: "expired", result: structuredClone(interaction.result) as unknown as Record<string, unknown> } });
continue;
}
const supersessionOutcome = interaction.status === "expired"
&& ["superseded_by_newer_request", "superseded_by_comment", "stale_target"].includes(String(interactionResult.outcome));
if (supersessionOutcome) {
const duplicate = interactionResult.outcome === "superseded_by_newer_request";
const decision = resolveNativeAttentionStatus({
facts: duplicate
? {
companyScopeValid: true,
responseState: "none",
route: "duplicate",
summary: interaction.summary ?? interaction.title ?? "Duplicate native attention request",
}
: {
companyScopeValid: true,
responseState: "stale",
route: "context",
summary: interaction.summary ?? interaction.title ?? "Stale native attention response",
},
priorIssueStatus: issue.status as NativeAuthoritativeIssueStatus,
agentId: input.agentId,
});
await applyNativeAttentionStatusDecision({
db: input.db,
companyId: input.companyId,
issueId: input.issueId,
runId: input.runId,
decision,
targetInteractionId: interaction.id,
canonicalRequestId: typeof interactionResult.supersededByInteractionId === "string"
? interactionResult.supersededByInteractionId
: undefined,
});
continue;
}
if (
interaction.resolvedByRunId === input.runId
|| (
interaction.resolvedByAgentId === input.agentId
&& interaction.createdByAgentId === input.agentId
)
) {
throw new NativeInteractionBridgeError(
"native_interaction_self_approval",
`Run ${input.runId} cannot consume a response it resolved itself`,
);
}
const hasInteractionResolver = Boolean(interaction.resolvedByUserId || interaction.resolvedByAgentId);
if (interaction.kind !== "request_item_verdicts" && !hasInteractionResolver) {
throw new NativeInteractionBridgeError(
"native_interaction_unresolved",
`Interaction ${interaction.id} has no authorized resolver`,
);
}
if (interaction.kind === "request_confirmation" || interaction.kind === "request_checkbox_confirmation") {
if (interaction.resolvedByAgentId === input.agentId) {
throw new NativeInteractionBridgeError(
"native_interaction_self_approval",
`Agent ${input.agentId} cannot consume a confirmation it resolved`,
);
}
if (
(interaction.status !== "accepted" && interaction.status !== "rejected")
|| !interaction.result
|| (interaction.result.outcome !== "accepted" && interaction.result.outcome !== "rejected")
) {
throw new NativeInteractionBridgeError(
"native_interaction_unresolved",
`Confirmation ${interaction.id} is not authoritatively resolved`,
);
}
const resolution = resolveNativeAttentionStatus({
facts: {
companyScopeValid: true,
responseState: "resolved",
route: "context",
summary: interaction.summary ?? interaction.title ?? "Resolved native interaction",
},
priorIssueStatus: issue.status as NativeAuthoritativeIssueStatus,
agentId: input.agentId,
});
if (resolution.reasonCode !== "attention_resolved_from_context") {
throw new NativeInteractionBridgeError("native_attention_resolution_invalid", "Resolved interaction has no continuation policy");
}
responses.push({
interactionId: interaction.id,
kind: interaction.kind,
response: {
status: interaction.status,
result: structuredClone(interaction.result) as unknown as Record<string, unknown>,
},
});
continue;
}
if (interaction.kind === "suggest_tasks" || interaction.kind === "connection_intent") {
if (
(interaction.status !== "accepted" && interaction.status !== "rejected")
|| !interaction.result
) {
throw new NativeInteractionBridgeError(
"native_interaction_unresolved",
`Suggested-task interaction ${interaction.id} is not authoritatively resolved`,
);
}
responses.push({
interactionId: interaction.id,
kind: interaction.kind,
response: {
status: interaction.status,
result: structuredClone(interaction.result) as unknown as Record<string, unknown>,
},
});
continue;
}
if (interaction.kind === "ask_user_questions") {
if (interaction.status !== "answered" || !interaction.result || interaction.result.cancelled === true) {
throw new NativeInteractionBridgeError(
"native_interaction_unresolved",
`Question interaction ${interaction.id} is not authoritatively answered`,
);
}
const resolution = resolveNativeAttentionStatus({
facts: {
companyScopeValid: true,
responseState: "resolved",
route: "context",
summary: interaction.summary ?? interaction.title ?? "Answered native interaction",
},
priorIssueStatus: issue.status as NativeAuthoritativeIssueStatus,
agentId: input.agentId,
});
if (resolution.reasonCode !== "attention_resolved_from_context") {
throw new NativeInteractionBridgeError("native_attention_resolution_invalid", "Answered interaction has no continuation policy");
}
const result = structuredClone(interaction.result) as unknown as Record<string, unknown>;
const summaryMarkdown = formatDurableQuestionResponseSummary(interaction);
responses.push({
interactionId: interaction.id,
kind: interaction.kind,
response: {
status: interaction.status,
result: summaryMarkdown
? { ...result, summaryMarkdown }
: result,
},
});
continue;
}
if (interaction.kind === "request_item_verdicts") {
const result = record(interaction.result);
const items = Array.isArray(result.items) ? result.items.map(record) : [];
const complete = result.complete === true;
const hasAuthorizedItemResolver = items.some((item) =>
typeof item.resolvedByUserId === "string"
|| (typeof item.resolvedByAgentId === "string" && item.resolvedByAgentId !== input.agentId));
const hasSelfResolvedItem = items.some((item) => item.resolvedByAgentId === input.agentId);
if (hasSelfResolvedItem) {
throw new NativeInteractionBridgeError(
"native_interaction_self_approval",
`Agent ${input.agentId} cannot consume item verdicts it resolved`,
);
}
if (
!interaction.result
|| (interaction.status !== "pending" && interaction.status !== "answered")
|| (interaction.status === "answered" && !complete)
|| items.length === 0
|| !hasAuthorizedItemResolver
) {
throw new NativeInteractionBridgeError(
"native_interaction_unresolved",
`Item-verdict interaction ${interaction.id} has no authoritative response`,
);
}
responses.push({
interactionId: interaction.id,
kind: interaction.kind,
response: {
status: interaction.status,
result: structuredClone(interaction.result) as unknown as Record<string, unknown>,
},
});
continue;
}
throw new NativeInteractionBridgeError(
"native_runtime_request_unsupported",
"Interaction kind is not supported by the native input contract",
);
}
const missing = [...requestedIds].filter((id) => !responses.some((response) => response.interactionId === id));
if (missing.length > 0) {
throw new NativeInteractionBridgeError(
"native_interaction_missing",
`Native interaction response not found: ${missing.join(", ")}`,
);
}
return responses.sort((left, right) => left.interactionId.localeCompare(right.interactionId));
}
export type LegacyQuestionResponseWakeProjection = {
interactionId: string;
summaryMarkdown: string;
};
/**
* Legacy adapters do not consume the runner's closed interaction-response
* envelope. Project an answered question set into their wake prompt at invoke
* time instead. The caller must keep this projection ephemeral so the
* interaction result remains the only control-plane copy of the answers.
*/
export async function materializeLegacyQuestionResponseWakeProjection(input: {
db: Db;
companyId: string;
issueId: string;
runId: string;
agentId: string;
interactionId: string;
}): Promise<LegacyQuestionResponseWakeProjection | null> {
const responses = await materializeNativeInteractionResponses({
...input,
interactionIds: [input.interactionId],
});
const response = responses.find((candidate) =>
candidate.interactionId === input.interactionId
&& candidate.kind === "ask_user_questions"
&& candidate.response.status === "answered"
);
if (!response) return null;
const summaryMarkdown = record(response.response.result).summaryMarkdown;
if (typeof summaryMarkdown !== "string" || summaryMarkdown.trim().length === 0) return null;
return {
interactionId: response.interactionId,
summaryMarkdown: summaryMarkdown.trim(),
};
}
/** Native runtimes never receive credentials and never auto-approve requests. */
export function rejectUnsupportedNativeRuntimeRequest(requestKind: string) {
return {
accepted: false as const,
code: "native_runtime_request_unsupported" as const,
requestKind,
credentialsInjected: false as const,
selfApproval: false as const,
};
}
export type NativeAttentionFacts = {
companyScopeValid: boolean;
responseState: "none" | "resolved" | "stale" | "expired";
route: "alternate_track" | "context" | "retry" | "agent" | "human" | "duplicate" | "recovery";
summary: string;
budgetExhausted?: boolean;
governanceGate?: NativeGovernanceGate | null;
resolvedTargetAgentId?: string | null;
};
export type NativeAttentionCandidate = {
route: NativeAttentionFacts["route"];
summary: string;
responseState?: NativeAttentionFacts["responseState"];
targetAgentId?: string | null;
targetCompanyId?: string | null;
budgetExhausted?: boolean;
governanceGate?: NativeGovernanceGate | null;
targetInteractionId?: string;
canonicalRequestId?: string;
};
function nonEmptyText(value: unknown) {
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
}
function candidateFromPersistedRequest(request: Record<string, unknown>): NativeAttentionCandidate {
const summary = nonEmptyText(request.summary);
if (!summary) {
throw new NativeInteractionBridgeError(
"native_attention_request_invalid",
"Persisted native attention requires a summary",
);
}
const target = record(request.target);
const capability = nonEmptyText(request.requestedCapability) ?? nonEmptyText(request.kind);
const ownerClass = nonEmptyText(target.ownerClass);
const requiredAuthority = nonEmptyText(request.requiredAuthority);
const responseState = nonEmptyText(request.responseState);
const governanceGate = record(request.governanceGate);
let route: NativeAttentionCandidate["route"];
if (capability === "context_lookup") route = "context";
else if (capability === "retry") route = "retry";
else if (capability === "duplicate") route = "duplicate";
else if (capability === "alternate_track" || ownerClass === "current_agent") route = "alternate_track";
else if (capability === "domain_expertise" || ownerClass === "agent") route = "agent";
else if (
["review", "credential_grant", "governed_approval", "subjective_decision", "external_action"].includes(capability ?? "")
|| ["board_user", "external_system"].includes(ownerClass ?? "")
|| ["board", "external"].includes(requiredAuthority ?? "")
) route = "human";
else {
throw new NativeInteractionBridgeError(
"native_attention_request_invalid",
`Persisted native attention capability ${capability ?? "unknown"} has no authorized route`,
);
}
return {
route,
summary,
responseState: ["none", "resolved", "stale", "expired"].includes(responseState ?? "")
? responseState as NativeAttentionFacts["responseState"]
: undefined,
targetAgentId: nonEmptyText(target.agentId),
targetCompanyId: nonEmptyText(target.companyId),
budgetExhausted: request.budgetExhausted === true,
governanceGate: governanceGate.kind === "interaction" && nonEmptyText(governanceGate.id)
? { kind: "interaction", id: nonEmptyText(governanceGate.id)! }
: null,
targetInteractionId: nonEmptyText(request.targetInteractionId) ?? undefined,
canonicalRequestId: nonEmptyText(request.canonicalRequestId) ?? undefined,
};
}
export type PersistedNativeAttentionReceipt = {
requestId: string;
route: NativeAttentionCandidate["route"];
assessmentId: string;
decisionId: string | null;
reasonCode: string | null;
resolvedTargetAgentId: string | null;
materializedTargets: NativeAttentionMaterializedTarget[];
replayed: boolean;
};
/**
* Runtime-reachable ingress for accepted attention facts. It starts from the
* immutable package result row, derives an authorized route, appends one
* assessment/decision lineage per request, and lets the existing issue and
* interaction owners materialize the effect.
*/
export async function routePersistedNativeResultAttention(input: {
db: Db;
runId: string;
}): Promise<PersistedNativeAttentionReceipt[]> {
const binding = await input.db.select({
companyId: heartbeatRuns.companyId,
runtimeMode: heartbeatRuns.runtimeMode,
resultId: nativeRunFinalizations.resultId,
issueId: nativeRunFinalizations.issueId,
}).from(heartbeatRuns)
.innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
.where(eq(heartbeatRuns.id, input.runId))
.limit(1)
.then((rows) => rows[0] ?? null);
if (!binding || binding.runtimeMode !== "native" || !binding.resultId) {
throw new NativeInteractionBridgeError(
"native_attention_binding_mismatch",
"Accepted native result binding not found",
);
}
const resultRow = await input.db.select().from(nativeRunResults).where(and(
eq(nativeRunResults.id, binding.resultId),
eq(nativeRunResults.companyId, binding.companyId),
eq(nativeRunResults.issueId, binding.issueId),
eq(nativeRunResults.runId, input.runId),
eq(nativeRunResults.schemaStatus, "accepted"),
)).limit(1).then((rows) => rows[0] ?? null);
if (!resultRow) {
throw new NativeInteractionBridgeError(
"native_attention_result_missing",
"Accepted native result row not found",
);
}
const result = record(record(resultRow.resultJson).result);
const requests = Array.isArray(result.attentionRequests)
? result.attentionRequests.map(record)
: [];
const receipts: PersistedNativeAttentionReceipt[] = [];
for (const [index, request] of requests.entries()) {
const candidate = candidateFromPersistedRequest(request);
const requestId = nonEmptyText(request.id) ?? nativeSha256({
runId: input.runId,
resultId: resultRow.id,
index,
request,
});
const existingAssessment = await input.db.select({ id: workAssessments.id })
.from(workAssessments).where(and(
eq(workAssessments.companyId, binding.companyId),
eq(workAssessments.issueId, binding.issueId),
eq(workAssessments.runId, input.runId),
eq(workAssessments.triggerKind, "native_attention"),
eq(workAssessments.triggerRef, requestId),
)).limit(1).then((rows) => rows[0] ?? null);
if (existingAssessment) {
const existingDecision = await input.db.select({
id: statusDecisions.id,
reasonCode: statusDecisions.reasonCode,
decisionJson: statusDecisions.decisionJson,
}).from(statusDecisions).where(and(
eq(statusDecisions.companyId, binding.companyId),
eq(statusDecisions.assessmentId, existingAssessment.id),
)).limit(1).then((rows) => rows[0] ?? null);
if (existingDecision) {
const decisionJson = record(existingDecision.decisionJson);
const effects = Array.isArray(decisionJson.effects) ? decisionJson.effects.map(record) : [];
const delegation = effects.find((effect) => effect.kind === "create_delegated_issue");
const finalizationError = effects.find((effect) => effect.kind === "record_finalization_error");
await input.db.transaction(async tx => {
await notifyNativeFinalizationWork(tx, { companyId: binding.companyId, runId: input.runId });
await tx.update(nativeRunFinalizations).set({
phase: finalizationError ? "retryable_failure" : "committed",
assessmentId: existingAssessment.id,
decisionId: existingDecision.id,
leaseOwner: null,
leaseExpiresAt: null,
recoveryState: null,
recoveryRequestId: null,
failureCode: nonEmptyText(finalizationError?.cause),
failureDetail: finalizationError ? {
originalFailureCode: finalizationError.cause,
nextAction: finalizationError.nextAction,
} : null,
nextAttemptAt: finalizationError ? new Date(Date.now() + 30_000) : null,
updatedAt: new Date(),
}).where(and(
eq(nativeRunFinalizations.runId, input.runId),
eq(nativeRunFinalizations.companyId, binding.companyId),
eq(nativeRunFinalizations.issueId, binding.issueId),
));
});
receipts.push({
requestId,
route: candidate.route,
assessmentId: existingAssessment.id,
decisionId: existingDecision.id,
reasonCode: existingDecision.reasonCode,
resolvedTargetAgentId: nonEmptyText(delegation?.agentId),
materializedTargets: [],
replayed: true,
});
continue;
}
}
const issue = await input.db.select({
status: issues.status,
statusVersion: issues.statusVersion,
lastStatusDecisionId: issues.lastStatusDecisionId,
}).from(issues).where(and(
eq(issues.id, binding.issueId),
eq(issues.companyId, binding.companyId),
)).limit(1).then((rows) => rows[0] ?? null);
if (!issue) throw new NativeInteractionBridgeError("native_attention_binding_mismatch", "Issue binding not found");
const coordinator = await input.db.select({
phase: nativeRunFinalizations.phase,
assessmentId: nativeRunFinalizations.assessmentId,
decisionId: nativeRunFinalizations.decisionId,
}).from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, input.runId))
.limit(1).then((rows) => rows[0] ?? null);
if (!coordinator) throw new NativeInteractionBridgeError("native_attention_binding_mismatch", "Coordinator binding not found");
const assessment = await recordNativeAttentionAssessment({
db: input.db,
companyId: binding.companyId,
issueId: binding.issueId,
runId: input.runId,
turnId: resultRow.turnId,
contractId: resultRow.completionContractId,
resultId: resultRow.id,
requestId,
request,
routingFacts: candidate as unknown as Record<string, unknown>,
priorIssueStatus: issue.status,
priorStatusVersion: Number(issue.statusVersion),
priorDecisionId: issue.lastStatusDecisionId,
supersedesAssessmentId: coordinator.assessmentId,
});
const routed = await routeNativeAttention({
db: input.db,
runId: input.runId,
assessmentId: assessment.id,
supersedesCommittedDecisionId: coordinator.phase === "committed"
? coordinator.decisionId ?? undefined
: undefined,
candidate,
});
receipts.push({
requestId,
route: candidate.route,
assessmentId: assessment.id,
decisionId: routed.decisionId,
reasonCode: routed.decision.reasonCode,
resolvedTargetAgentId: routed.resolvedTargetAgentId,
materializedTargets: routed.targets,
replayed: existingAssessment !== null,
});
}
if (receipts.length > 0) {
await input.db.transaction(async (tx) => {
const run = await tx.select({ resultJson: heartbeatRuns.resultJson })
.from(heartbeatRuns).where(and(
eq(heartbeatRuns.id, input.runId),
eq(heartbeatRuns.companyId, binding.companyId),
)).for("update").limit(1).then((rows) => rows[0] ?? null);
if (!run) throw new NativeInteractionBridgeError("native_attention_binding_mismatch", "Heartbeat run missing");
await tx.update(heartbeatRuns).set({
resultJson: { ...record(run.resultJson), nativeAttentionRouting: receipts },
updatedAt: new Date(),
}).where(eq(heartbeatRuns.id, input.runId));
});
}
return receipts;
}
/**
* Internal owner-routing step for native attention. The persisted-result
* ingress above supplies the server-owned assessment and untrusted candidate;
* this step resolves company scope and commits or audit-materializes the
* authority decision.
*/
export async function routeNativeAttention(input: {
db: Db;
runId: string;
assessmentId?: string;
supersedesCommittedDecisionId?: string;
candidate: NativeAttentionCandidate;
}) {
const binding = await input.db.select({
companyId: heartbeatRuns.companyId,
agentId: heartbeatRuns.agentId,
runtimeMode: heartbeatRuns.runtimeMode,
issueId: nativeRunFinalizations.issueId,
assessmentId: nativeRunFinalizations.assessmentId,
issueStatus: issues.status,
issueStatusVersion: issues.statusVersion,
issueDecisionId: issues.lastStatusDecisionId,
}).from(heartbeatRuns)
.innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
.innerJoin(issues, and(
eq(issues.id, nativeRunFinalizations.issueId),
eq(issues.companyId, heartbeatRuns.companyId),
))
.where(eq(heartbeatRuns.id, input.runId))
.limit(1)
.then((rows) => rows[0] ?? null);
if (!binding || binding.runtimeMode !== "native") {
throw new NativeInteractionBridgeError("native_attention_binding_mismatch", "Native attention run binding not found");
}
const companyAgents = await input.db.select({
id: agents.id,
companyId: agents.companyId,
name: agents.name,
status: agents.status,
reportsTo: agents.reportsTo,
}).from(agents).where(eq(agents.companyId, binding.companyId)).orderBy(asc(agents.createdAt), asc(agents.id));
const targetSuggestion = input.candidate.targetAgentId
? await input.db.select({ id: agents.id, companyId: agents.companyId })
.from(agents).where(eq(agents.id, input.candidate.targetAgentId)).limit(1)
.then((rows) => rows[0] ?? null)
: null;
const companyScopeValid = (
!input.candidate.targetCompanyId
|| input.candidate.targetCompanyId === binding.companyId
) && (
!input.candidate.targetAgentId
|| targetSuggestion?.companyId === binding.companyId
);
let resolvedTargetAgentId: string | null = null;
if (companyScopeValid && input.candidate.route === "agent") {
const eligible = companyAgents.filter((candidate) =>
candidate.id !== binding.agentId
&& getAgentWorkEligibility({ agent: candidate, agents: companyAgents }).invokable
);
const requested = eligible.find((candidate) => candidate.id === input.candidate.targetAgentId);
resolvedTargetAgentId = requested?.id ?? eligible[0]?.id ?? null;
}
const facts: NativeAttentionFacts = {
companyScopeValid,
responseState: input.candidate.responseState ?? "none",
route: input.candidate.route === "agent" && !resolvedTargetAgentId
? "recovery"
: input.candidate.route,
summary: input.candidate.summary,
budgetExhausted: input.candidate.budgetExhausted,
governanceGate: input.candidate.governanceGate,
resolvedTargetAgentId,
};
const decision = resolveNativeAttentionStatus({
facts,
priorIssueStatus: binding.issueStatus as NativeAuthoritativeIssueStatus,
agentId: binding.agentId,
});
const auditOnly = decision.reasonCode === "attention_duplicate_suppressed";
if (auditOnly) {
const targets = await applyNativeAttentionStatusDecision({
db: input.db,
companyId: binding.companyId,
issueId: binding.issueId,
runId: input.runId,
decision,
targetInteractionId: input.candidate.targetInteractionId,
canonicalRequestId: input.candidate.canonicalRequestId,
});
return { decision, decisionId: null, targets, resolvedTargetAgentId };
}
const assessmentId = input.assessmentId ?? binding.assessmentId;
if (!assessmentId) {
throw new NativeInteractionBridgeError(
"native_attention_assessment_missing",
"Native attention requires the server-owned work assessment for this result",
);
}
const committed = await commitNativeStatusDecision({
db: input.db,
companyId: binding.companyId,
issueId: binding.issueId,
runId: input.runId,
assessmentId,
priorStatus: binding.issueStatus,
priorStatusVersion: Number(binding.issueStatusVersion),
priorDecisionId: binding.issueDecisionId,
decision,
supersedesCommittedDecisionId: input.supersedesCommittedDecisionId,
});
return {
decision,
decisionId: committed.decision.id,
targets: [],
resolvedTargetAgentId,
};
}
/** Server-owned attention resolution from canonical routing facts. */
export function resolveNativeAttentionStatus(input: {
facts: NativeAttentionFacts;
priorIssueStatus: NativeAuthoritativeIssueStatus;
agentId: string;
}): NativeStatusDecision {
const preserve = (reasonCode: string, effects: NativeStatusEffect[]): NativeStatusDecision => ({
policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
statusAction: "preserve",
toStatus: input.priorIssueStatus,
reasonCode,
unblockDescriptor: null,
effects,
});
const continuation = (reasonCode: string): NativeStatusDecision => ({
policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
statusAction: "in_progress",
toStatus: "in_progress",
reasonCode,
unblockDescriptor: null,
effects: [{
kind: "enqueue_continuation",
continuationKind: input.facts.route === "agent" ? "response_wake" : "same_agent",
summary: input.facts.summary,
idempotencyKey: `native-attention:${input.facts.route}:${input.facts.summary}`,
agentId: input.agentId,
}],
});
if (!input.facts.companyScopeValid) {
return preserve("result_schema_rejected", [{
kind: "record_finalization_error",
cause: "result_schema_rejected",
nextAction: "Remove the cross-company attention target.",
agentId: input.agentId,
}]);
}
if (input.facts.responseState === "stale") {
return preserve("attention_duplicate_suppressed", [{ kind: "record_stale_response" }]);
}
if (input.facts.responseState === "expired") {
return preserve("attention_budget_exhausted", [{ kind: "record_expiry" }]);
}
if (input.facts.budgetExhausted || input.facts.route === "recovery") {
return preserve("attention_budget_exhausted", [{
kind: "record_finalization_error",
cause: "attention_budget_exhausted",
nextAction: "Assign a named recovery owner.",
agentId: input.agentId,
}]);
}
if (input.facts.route === "duplicate") {
return preserve("attention_duplicate_suppressed", [{ kind: "link_canonical_request" }]);
}
if (input.facts.governanceGate) {
return {
policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
statusAction: "in_review",
toStatus: "in_review",
reasonCode: "governed_gate_pending",
unblockDescriptor: null,
effects: [
{ kind: "create_interaction", gate: input.facts.governanceGate },
{ kind: "notify_owner", agentId: input.agentId, reason: "governed_gate_pending" },
],
};
}
if (input.facts.route === "alternate_track") {
return continuation("turn_waiting_other_track_live");
}
if (input.facts.route === "context" || input.facts.route === "retry") {
return continuation("attention_resolved_from_context");
}
if (input.facts.route === "agent") {
if (!input.facts.resolvedTargetAgentId) {
return preserve("attention_budget_exhausted", [{
kind: "record_finalization_error",
cause: "attention_no_valid_route",
nextAction: "Assign a same-company eligible delegate.",
agentId: input.agentId,
}]);
}
const decision = continuation("attention_routed_to_agent");
return {
...decision,
effects: [
{ kind: "create_delegated_issue", agentId: input.facts.resolvedTargetAgentId, summary: input.facts.summary },
...decision.effects,
],
};
}
if (input.facts.route === "human") {
return {
policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
statusAction: "in_review",
toStatus: "in_review",
reasonCode: "attention_requires_human_authority",
unblockDescriptor: null,
effects: [
{ kind: "create_interaction", prompt: input.facts.summary },
{ kind: "notify_owner", agentId: input.agentId, reason: "attention_requires_human_authority" },
],
};
}
throw new NativeInteractionBridgeError("native_attention_facts_invalid", "Attention facts do not select a route");
}