mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-20 06:02:18 +02:00
fix(grok): avoid routine wakes for accepted work (#16394)
## Thinking Path
> - Paperclip lets people manage AI agents and their work.
> - Grok Bot receives assignments through routine webhooks and uses
agent MCP tools.
> - Each accepted assignment belongs to the current external
interaction.
> - A pending tool result or comment could start another routine while
that work was still active.
> - Quiet MCP traffic cannot prove that Grok has stopped working.
> - This pull request keeps receipts and feedback durable for the
current interaction to read.
> - New task routines start only for offers that the Bot has not
accepted.
## Linked Issues or Issue Description
Refs #16219 and #16305. The duplicate search also found #12991, which
covers instruction files for local Grok runs.
**What happened?**
The dispatcher could start another routine for an unread result after a
period without a routine reply. It could also wake accepted work for a
task comment. A queued delivery could still start after the Bot had
claimed the offer.
**Expected behavior**
Accepted work stays in its current interaction. The Bot polls the
original operation receipt and reads queued feedback at a safe boundary.
Acknowledgements of a sent wake remain valid after acceptance.
**Steps to reproduce**
1. Accept a Grok assignment and leave it active.
2. Add an unread comment, or complete a tool operation whose call
returned pending.
3. Run the dispatcher after its quiet period.
4. Observe another webhook call. Also queue an offer wake, accept the
offer before sending, and observe that delivery still starts.
**Paperclip version or commit**
Observed during real-client qualification on Core
ff0f976de1. The three new regression cases
also failed on current master before the correction.
## What Changed
- Stop automatic result and feedback wakes for accepted work. Keep both
available through their existing read paths.
- Require an unaccepted offer before dispatch or an explicit delivery
retry. Serialize the final HTTP handoff with acceptance receipts. Cancel
obsolete unsent wakes. Preserve late acknowledgements and immediate Stop
authority.
- Return current Grok polling instructions on assignment reads. Clarify
pending and unknown outcomes in setup guidance.
- Add protocol, receipt-polling, queued-delivery and
late-acknowledgement regressions. Update the operation guide and
readiness plan.
## Verification
- Three new regression cases fail before the fix because a webhook
starts.
- The latest 31 Grok protocol checks and 47 existing Dot Runner checks
pass. New tests force acceptance during secret lookup and the final send
check. They also prove that acknowledgements and Stop remain free during
pending delivery.
- The final-head Grok, Dot, MCP and question-delivery suite passes all
212 checks across six files. Dedicated onboarding UI checks passed six.
- Dedicated Runner driver checks use the real local Rust daemon. Built
the debug daemon required by that package; all 17 checks then passed.
- Repository typecheck and build pass. Token gates, the 2,731-story
structural catalog and the Storybook build pass.
- Public synthetic marker for the post-deployment native source-read
check: `grok-native-read-b45be4eff22cb15a31d0af512f437b8d`.
- Final-head CI passes all 55 checks, with two optional Storybook skips.
Greptile is 5/5 with no open threads. The first chat-shard attempt
failed one Telegram SQL-quiet timing assertion; all five related cases
pass locally and the single targeted CI rerun passes.
- After deployment, repeat actual Grok work through an answered
question, receipt polling, a saved report and file, canonical completion
and cleanup. This correction is not production qualification by itself.
## Risks
A Bot that does not poll a pending receipt can wait without another
routine wake. Current assignment reads now supply polling instructions
to existing connections. Task comments stay queued until the Bot reads
them or Runner admits the next continuation. The real client must verify
that behavior before production enablement. The observed second file
version does not prove which provider interaction created it. Acceptance
can wait for the bounded webhook response. Stop and authenticated
acknowledgements use independent row locks and remain available.
## Model Used
OpenAI GPT-6 in Codex, with reasoning, repository editing, shell tools
and browser tools. The exact serving model ID, snapshot and context
window are not exposed by this session.
## 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>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
854ab294cc
commit
074c67fea8
8 files changed
+266
-81
No files matched your search
@@ -49,6 +49,23 @@ completion verifies that receipt against the upload and admitted assignment.
|
||||
A visible attachment without that verified publication does not satisfy a task
|
||||
that requires a file output.
|
||||
|
||||
## Feedback and pending operations
|
||||
|
||||
An assignment wake starts a Grok routine. Once the Bot accepts that work,
|
||||
Paperclip keeps new comments and asynchronous operation receipts durable without
|
||||
starting another routine for that session. Silence does not mean the Bot is idle.
|
||||
Acceptance and the final webhook handoff are serialized. If acceptance finishes
|
||||
while credentials are loading, the unsent wake is cancelled. Stop and
|
||||
authenticated acknowledgements remain independent of a pending HTTP response.
|
||||
Human-question answers and resolved durable waits resume through normal Runner
|
||||
admission and a new assignment wake.
|
||||
|
||||
If a tool returns pending, poll `paperclip_grok_operation_status` with the same
|
||||
assignment and request IDs for its durable outcome. Do not repeat it under a new
|
||||
ID or perform dependent work first. An unknown outcome requires reconciliation. At safe boundaries and before finishing,
|
||||
the Bot reads its inbox and current task history for queued feedback. Current
|
||||
assignment reads include these instructions, including for existing connections.
|
||||
|
||||
## Recovery and stop
|
||||
|
||||
Inspect a rejected delivery before using **Retry rejected delivery**. An uncertain
|
||||
|
||||
@@ -1,6 +1,35 @@
|
||||
# Grok Bot production readiness
|
||||
|
||||
Status: implementation and qualification in progress. Actual client OAuth and a routine wake with an exact chat acknowledgement are observed. The candidate has local authenticated challenge receipts, real Rust assigned-work/file/finalization tests, and durable proactive-intake tests. Deployed two-way work acceptance remains a release gate.
|
||||
Status as of October 15: a real assigned task now reaches canonical completion. Qualification found an active-session wake hazard; its correction is implemented and passing local checks. Production acceptance is incomplete. The current release goal and matrix below supersede the earlier dated campaign entries. Historical successes and failures remain evidence for the revisions actually tested.
|
||||
|
||||
## Current production finish line — October 14
|
||||
|
||||
Ship Grok Bot as a user-selectable, opt-in experimental agent in production Paperclip Cloud. A user can connect an existing Bot, assign ordinary tasks and receive questions and useful outputs in Paperclip; the Bot can independently initiate governed work. The user selected a dedicated synthetic Grok Qualification production workspace, which has been created without an agent grant or smoke task.
|
||||
|
||||
There is no testing call, case-count, token or spending cap. The former six-case/twenty-webhook limit is withdrawn by the user. Test as needed to meet the finish conditions. Normal company permissions, recorded budgets, admission, delivery reconciliation and resource cleanup remain required; Grok spending is unknown.
|
||||
|
||||
| Finish condition | Evidence required | Current outcome / next action |
|
||||
| --- | --- | --- |
|
||||
| Ordinary onboarding works | Actual Grok UI, fixed agent consent, saved setup, wrong-key behavior, authenticated challenge, refresh, reconnect and revocation | Resumed real Grok authorization, saved setup Done, authenticated routine challenge and automatic refresh passed on the October 15 serving revision. Runtime reconnect copy fix is merged; verify its published UI and fresh production setup next. |
|
||||
| Assigned work reaches a usable result | One actual managed workflow: progress, question and answer, document, downloaded file/hash, canonical completion or review and cleanup | A preliminary comparison task on serving Core ff0 completed progress → question → answer → report/CSV → canonical Done, with both cleanup receipts. The cited CSV downloads with the exact publication hash. Two distinct CSV uploads accompanied an active-session result wake; correct that wake behavior and requalify the final candidate. |
|
||||
| Bot initiates governed work | Real-client task creation/claim/completion with stable receipts and no outbound echo or duplicate effects | Unverified. Run after the assigned-work gate. |
|
||||
| Recovery and authority are reliable | Healthy work past the prior five-minute failure window; actual controller restart preserving assignment/session/process and no duplicate wake; Stop, pause, reassignment, revocation, stale calls, queues and cleanup | The old hold was inspected and cleared. The successful comparison continuation ended below five minutes. Healthy work beyond that window, an actual restart and active Stop remain unverified; protocol tests are separate evidence. |
|
||||
| One identifiable reviewed candidate passes source gates | Focused protocol/auth/Runner/UI tests, full CI/typecheck/tests/build/token/Storybook evidence and exact artifacts | The active-session wake correction follows origin/master. Initial local checks used d6e4b36c8c049b85813caf3e3cbf1f8d87e4e068; CI and review must verify the final PR head. Three regressions failed before the fix. Focused tests, repository typecheck/build, token gates and Storybook checks pass. Full CI, review and publication of this correction remain required; prior candidate receipts are historical. |
|
||||
| Production opt-in is delivered and smoke-tested | Staging acceptance, correct served Core/distribution/Runner, normal editable opt-in controls/defaults, actual production Bot workflow and inspectable task/output links | Cloud opt-in policy and release-guide corrections are merged. Staging Cloud provenance and minimal feature values were verified, with Dot disabled and defaults unchanged. The corrected wake candidate, remaining real-client cases, ordinary production opt-in controls and production smoke are outstanding. |
|
||||
|
||||
The critical path is to review and publish the active-session wake correction, verify its served identity, then complete the remaining real-client acceptance cases. The selected staging workspace remains the test destination. Preserve one source manifest per case; do not combine incompatible runs into one green release.
|
||||
|
||||
Real Grok reconnect restored the same agent, plugin and webhook-only routine. Saved setup reached Done after both delivery and authenticated MCP acknowledgement. After the initial access token expired, Grok refreshed it and acknowledged another challenge without reauthorization. Grok/MCP/Runner were enabled with Dot disabled; setup allocated no managed Runner. The operator used the existing saved-invitation path to get the secure pairing-code copy control before the Runtime copy correction published. Record that as assisted recovery, not proof of unassisted fresh onboarding. The copy correction is merged in #16373 after 55 passing CI checks, two optional skips and review 5/5.
|
||||
|
||||
The preliminary comparison task ran on serving Core ff0f976de1036555fb9deda8c1f31a5f748723f6. Grok accepted it, posted progress, asked a question, resumed after the synthetic Solo builder answer and saved a 284-word report with official links. Canonical completion accepted the authenticated file-publication evidence and set the task to Done. Both managed runs committed and both leases have successful cleanup receipts. The cited CSV opens through the normal download link and matches its recorded 1,277-byte hash, three rows and five columns. No chat nudge or manual task disposition was used. The continuation ended after about 280 seconds, below the prior five-minute failure window.
|
||||
|
||||
An unread asynchronous tool result also caused a webhook during accepted work. Two CSV versions with different request IDs and hashes appeared. Their cause cannot be established without the provider transcript, but quiet MCP traffic cannot establish provider idleness. The correction removes receipt and accepted-feedback self-wakes, retains durable polling and inbox/history reads, and cancels queued dispatches whose offer was already accepted. Late acknowledgements still work. Assignment reads supply current polling guidance to existing connections. Three regressions failed before the correction. Review then identified acceptance during the last send check and during credential resolution. The final handoff now shares an advisory lock with Grok acceptance receipts; it holds no binding or assignment row lock across HTTP. Obsolete unsent wakes are cancelled. The 31 Grok protocol checks and 47 existing Dot Runner checks pass, including forced acceptance races, authenticated replies and Stop during pending delivery. Earlier broader protocol, onboarding, real-daemon, repository typecheck/build, token and Storybook checks passed; final-head CI remains required. These are source checks, not actual-provider requalification.
|
||||
|
||||
Core d73d7a8651db0897cb50adbca42731bcb8aed550 and its canonical private Cloud composition passed publication. One single-workspace request passed preparation but was blocked by an existing shared rollout; it created no campaign. The existing rollout later released the reservation. That prepared source is now superseded by the wake correction. Keep its receipts as history and do not resume its older deployment request. Select a new immutable reviewed source after this fix passes CI and review. Production opt-in policy and release-guide changes are merged, but production rollout and smoke remain outstanding.
|
||||
|
||||
Desktop control stopped responding after the successful exchanges. Resetting the tool did not restore it, and the user has been asked to reopen Codex. Browser control and webhook/MCP work remain usable; native Bot-initiated prompts and fresh production setup require restored desktop control or ordinary human setup actions. The dedicated production workspace has no qualification agent grant or smoke task.
|
||||
|
||||
Keep native providers and person-purpose MCP connections unchanged. No browser automation belongs in normal operation. Defer localhost relays, internal Grok APIs, marketplace distribution and Paperclip-connected third-party tools. Completion of a Paperclip work session does not assert remote Grok termination.
|
||||
|
||||
## Candidate and release contract
|
||||
|
||||
@@ -23,7 +52,7 @@ Use `.agents/skills/create-external-bot-integration/SKILL.md` and its release, R
|
||||
| Stop | Revoke Paperclip work authority and future dispatch; account for managed Runner cleanup; remote Grok activity remains unknown | “Stop Paperclip work” and “Open Grok”; no claim of global Bot termination |
|
||||
| Economics | Recorded budgets gate admission; Grok usage/spend unknown; managed Runner cost separate | Never report zero Grok cost or promise control over its billing |
|
||||
| Distribution | Experimental private connection on authenticated HTTPS instances and Cloud | Localhost relays, internal Grok APIs and public marketplace distribution remain deferred |
|
||||
| Qualification bounds | One active Bot assignment; short synthetic tasks; no paid soak; bounded attempts and explicit resource ownership | Start with at most six principal task cases and twenty webhook invocations; inspect ambiguity before another send; pause to review any additional spend or scope |
|
||||
| Qualification scope | One active Bot assignment, synthetic data, explicit resource ownership and inspection before repeating uncertain delivery | No arbitrary call, case-count or spending cap. Run the coverage needed for the agreed production finish line; preserve normal governance and cleanup. |
|
||||
|
||||
## Architecture decision
|
||||
|
||||
@@ -53,7 +82,7 @@ Use current lease-renewal and durable waiting paths. Distinguish pairing-code, a
|
||||
| P5 — review candidate | Focused auth/protocol/Runner/UI tests, full required checks, maps/token/Storybook checks and PR review green on one frozen candidate; compatible packaged Runner artifact | Codex; PR template, exact head/checks and immutable artifact digest |
|
||||
| P6 — deployed acceptance | Selected staging tenant serves candidate app, ingress and ordinary configured managed Runner image; fresh onboarding and both work directions pass without chat nudges or manual repair; cleanup accounted for | Codex; actual qualification Bot campaign, screenshots/private evidence, run/document/artifact links and provider deletion receipts |
|
||||
|
||||
The next critical gate is the remainder of P1. Scoped staging prerequisites, a temporary agent grant, the new Bot and its webhook-only qualification routine were explicitly approved. The temporary shared Dot gateway is client interoperability evidence only; it is not a production Grok provider or a passed Runner acceptance test. Keep the endpoint/data alive while the user/Bot tests it, and announce any intentional shutdown with resumption instructions.
|
||||
At this historical point, the next critical gate was the remainder of P1. Scoped staging prerequisites, a temporary agent grant, the new Bot and its webhook-only qualification routine were explicitly approved. The temporary shared Dot gateway is client interoperability evidence only; it is not a production Grok provider or a passed Runner acceptance test. Keep the endpoint/data alive while the user/Bot tests it, and announce any intentional shutdown with resumption instructions.
|
||||
|
||||
## Acceptance campaign
|
||||
|
||||
@@ -76,7 +105,7 @@ Before review-ready handoff run `pnpm -r typecheck`, `pnpm test:run`, `pnpm buil
|
||||
|
||||
Inspect current private Cloud runbooks and serving versions. Reuse existing tenant `/mcp/runner` routing first; change ingress only for a demonstrated missing route/auth behavior. Keep private Cloud source and qualification secrets out of public plans/PRs. Record the app image, ingress SHA, Runner digest, configured managed environment and resource receipts. A local binary override is not proof of the shipped managed image.
|
||||
|
||||
Production ready means one reviewed candidate, required checks green, the named staging deployment and shipped Runner verified, the ordinary C01/C02/C03 journeys witnessed with the approved qualification Bot, required recovery/capability cases passed, and all resources released or intentionally retained with an owner and resume path. Report review-ready, merged, staged and canary-verified independently. Fleet-default enablement and a production-wide rollout require their own explicit authorization.
|
||||
Production ready means one reviewed candidate, required checks green, the named staging deployment and shipped Runner verified, the ordinary C01/C02/C03 journeys witnessed with the approved qualification Bot, required recovery/capability cases passed, and all resources released or intentionally retained with an owner and resume path. Report review-ready, merged, staged and canary-verified independently. The user later selected the production opt-in release goal above. It authorizes the reviewed release and policy delivery after staging acceptance; globally enabling Grok by default remains outside scope.
|
||||
|
||||
## Provider sources checked October 13
|
||||
|
||||
@@ -119,7 +148,7 @@ The existing Dot readiness contract requires its authenticated MCP Events callba
|
||||
| Continued authorization | Capabilities call succeeded after the initial 15-minute token lifetime without a new consent step | Supports client renewal; token endpoint/rotation receipt not independently inspected |
|
||||
| Revocation | Paperclip revoked the temporary connection; real stale inbox attempt returned authentication-required and no data | Current connection boundary only; no task mutation/reassignment test yet |
|
||||
| Reconnect | Fresh one-use invitation and consent for the same agent; real read returned a replacement binding and empty inbox | Existing plugin reused; old binding revoked; operator-driven test |
|
||||
| Routine wake | Vendor-confirmed desktop row-click regression; Bot-generated Webhook URL navigation link opened the panel. One guarded invocation returned HTTP 200 and the exact nonce reply | Operator workaround required; chat acknowledgement is not a persisted authenticated nonce receipt in Paperclip. One of twenty permitted webhook calls used |
|
||||
| Routine wake | Vendor-confirmed desktop row-click regression; Bot-generated Webhook URL navigation link opened the panel. One guarded invocation returned HTTP 200 and the exact nonce reply | Operator workaround required; chat acknowledgement is not a persisted authenticated nonce receipt in Paperclip. One invocation observed under the historical cap, which the user later removed |
|
||||
| Current source checks | Local authenticated MCP challenge, Rust task → document/file → finalization, proactive intake/deduplication, and renewed-session recovery tests pass. Repository typecheck, build, token gates, maps and Storybook build pass | Source candidate is not deployed; complete final suite and candidate-bound real managed acceptance remain required |
|
||||
|
||||
Do not promote the temporary shared Dot connection to ready or substitute its event subscription with an unverified Grok webhook. The remaining P1 gate is the candidate’s durable authenticated challenge acknowledgement through the actual Grok client. Assigned work, idle intake, restart and managed cleanup still require real deployed acceptance. Local tests use scripted clients and synthetic storage; they do not count as provider acceptance.
|
||||
|
||||
@@ -4,4 +4,4 @@ export const EXTERNAL_AGENT_TOOL_GUIDANCE = `When you receive a Paperclip assign
|
||||
/** The assignment catalog constrains the Paperclip bridge, not the external runtime. */
|
||||
export const DOT_AGENT_TOOL_GUIDANCE = `${EXTERNAL_AGENT_TOOL_GUIDANCE}\n\nUse the assignment catalog through paperclip_dot_tool for Paperclip operations. Use your native tools directly for the rest of the work; they do not need to appear in that catalog. Record useful results and deliverables on the task and follow the Paperclip completion instructions.`;
|
||||
|
||||
export const GROK_BOT_AGENT_TOOL_GUIDANCE = `${EXTERNAL_AGENT_TOOL_GUIDANCE}\n\nUse the current assignment catalog through paperclip_grok_tool for first-party Paperclip operations. Grok Bot retains its own connected apps and persistent memory. Retrieve current task context and instructions for each Paperclip work session; do not assume provider memory is isolated or reset. Acknowledge routine wakes using paperclip_grok_ack_wake. Accept admitted assignments before task operations and keep request IDs stable across retries. Report progress, questions and deliverables in Paperclip, then follow its canonical completion instructions. Routine acceptance, Paperclip session completion and remote process termination are distinct. Grok usage and spending are not reported.`;
|
||||
export const GROK_BOT_AGENT_TOOL_GUIDANCE = `${EXTERNAL_AGENT_TOOL_GUIDANCE}\n\nUse the current assignment catalog through paperclip_grok_tool for first-party Paperclip operations. Grok Bot retains its own connected apps and persistent memory. Retrieve current task context and instructions for each Paperclip work session; do not assume provider memory is isolated or reset. Acknowledge routine wakes using paperclip_grok_ack_wake. Accept admitted assignments before task operations and keep request IDs stable across retries. If an operation is pending, poll paperclip_grok_operation_status with the same assignmentId and requestId for its durable outcome before dependent work. Completed or rejected receipts are terminal; an unknown outcome needs reconciliation. Do not repeat it under a new ID. A tool result does not start another routine. At safe boundaries and before finishing, drain paperclip_grok_inbox and read current task history to incorporate queued feedback. Report progress, questions and deliverables in Paperclip, then follow its canonical completion instructions. Routine acceptance, Paperclip session completion and remote process termination are distinct. Grok usage and spending are not reported.`;
|
||||
@@ -6,9 +6,9 @@ import { createPublicMcpExecutor } from "../services/public-mcp/capabilities.js"
|
||||
import { createDotRunnerMcpTools } from "../services/dot-runner-broker.js";
|
||||
import { createHash, randomBytes, randomUUID } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { beforeAll, afterAll, describe, expect, it, vi } from "vitest";
|
||||
import { beforeAll, afterAll, afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { agents, authUsers, companies, companyMemberships, createDb, dotAgentBindings, grokBotDispatches,
|
||||
grokBotRoutines, mcpOauthGrants, activityLog, issues, heartbeatRuns, dotRunnerAssignments, dotMailboxItems, nativeRunFinalizations } from "@paperclipai/db";
|
||||
grokBotRoutines, mcpOauthGrants, activityLog, issues, heartbeatRuns, dotRunnerAssignments, dotRunnerOperations, dotMailboxItems, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { accessService } from "../services/access.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
@@ -16,6 +16,8 @@ import { createPublicMcpOAuth } from "../services/public-mcp/oauth.js";
|
||||
import { dotRunnerBroker } from "../services/dot-runner-broker.js";
|
||||
import { createGrokRoutineDispatcher } from "../services/grok-bot/dispatcher.js";
|
||||
import { grokRoutineService } from "../services/grok-bot/routine.js";
|
||||
import * as secretModule from "../services/secrets.js";
|
||||
import * as externalAuthority from "../services/external-bot-authority.js";
|
||||
import { type GrokRoutineWake } from "../services/grok-bot/webhook.js";
|
||||
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
@@ -29,6 +31,8 @@ beforeAll(async () => {
|
||||
}, 120000);
|
||||
afterAll(async () => { try { await temporary?.cleanup(); } finally { vi.unstubAllEnvs(); } });
|
||||
|
||||
afterEach(() => vi.restoreAllMocks());
|
||||
|
||||
async function fixture() {
|
||||
const userId = randomUUID();
|
||||
await db.insert(authUsers).values({ id: userId, name: "Grok operator", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
|
||||
@@ -62,6 +66,37 @@ function acknowledgement(row: Awaited<ReturnType<typeof verification>>) {
|
||||
return { correlationId: row.id, bindingId: row.bindingId, bindingGeneration: row.bindingGeneration, challenge: wake.challenge };
|
||||
}
|
||||
|
||||
async function admittedWork(f: Awaited<ReturnType<typeof fixture>>, status: "offered" | "accepted") {
|
||||
const verified = await verification(f);
|
||||
await f.service.send(f.company.id, f.agent.id, verified.id);
|
||||
await f.service.acknowledge(f.principal, acknowledgement(verified));
|
||||
const sessionId = randomUUID();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id,
|
||||
status: "running", runtimeMode: "native", nativeSessionId: sessionId, invocationSource: "assignment", triggerDetail: "system" }).returning();
|
||||
const [task] = await db.insert(issues).values({ companyId: f.company.id, title: "Grok assignment",
|
||||
assigneeAgentId: f.agent.id, executionRunId: run!.id, checkoutRunId: run!.id, status: "in_progress" }).returning();
|
||||
await db.update(heartbeatRuns).set({ nativeIssueId: task!.id }).where(eq(heartbeatRuns.id, run!.id));
|
||||
await db.insert(nativeRunFinalizations).values({ runId: run!.id, companyId: f.company.id, issueId: task!.id,
|
||||
phase: "observed", controllerGeneration: 1, leaseExpiresAt: new Date(Date.now() + 60_000) });
|
||||
const [work] = await db.insert(dotRunnerAssignments).values({ companyId: f.company.id, agentId: f.agent.id,
|
||||
bindingId: f.pairing.bindingId, bindingGeneration: 1, runId: run!.id, normalizedSessionId: sessionId, turnId: randomUUID(),
|
||||
controllerGeneration: 1, catalogDigest: "test", status, projection: {},
|
||||
acceptBy: new Date(Date.now() + 60_000), expiresAt: new Date(Date.now() + 60_000) }).returning();
|
||||
return { run: run!, task: task!, work: work! };
|
||||
}
|
||||
|
||||
async function acceptance(f: Awaited<ReturnType<typeof fixture>>, work: typeof dotRunnerAssignments.$inferSelect) {
|
||||
const requestId = randomUUID();
|
||||
await db.insert(dotRunnerOperations).values({ companyId: f.company.id, assignmentId: work.id, requestId,
|
||||
digest: "accept", command: { action: "accept" }, status: "admitted" });
|
||||
const binding = { companyId: f.company.id, agentId: f.agent.id, bindingId: f.pairing.bindingId,
|
||||
bindingGeneration: 1, runId: work.runId, normalizedSessionId: work.normalizedSessionId,
|
||||
turnId: work.turnId, assignmentRevision: work.revision };
|
||||
const port = dotRunnerBroker(db).port({ binding, provider: { binding: { ...binding,
|
||||
acceptByUnixMs: work.acceptBy.getTime(), expiresAtUnixMs: work.expiresAt.getTime() } } });
|
||||
return () => port.settle({ sourceEventId: randomUUID(), payload: { requestId, binding, outcome: { status: "accepted" } } });
|
||||
}
|
||||
|
||||
describe("durable Grok routine qualification", () => {
|
||||
it("receives an authenticated Grok acknowledgement through the actual MCP transport without exposing personal or Dot event tools", async () => {
|
||||
const f = await fixture(); const row = await verification(f);
|
||||
@@ -230,25 +265,125 @@ describe("durable Grok routine qualification", () => {
|
||||
expect(await f.service.state(f.company.id, f.agent.id, row.bindingId)).toMatchObject({ verified: true });
|
||||
});
|
||||
|
||||
it("keeps rejected unread feedback available for an explicit retry without automatic resend", async () => {
|
||||
const f = await fixture(); const verified = await verification(f);
|
||||
await f.service.send(f.company.id, f.agent.id, verified.id);
|
||||
await f.service.acknowledge(f.principal, acknowledgement(verified));
|
||||
const sessionId = randomUUID();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id,
|
||||
status: "running", runtimeMode: "native", nativeSessionId: sessionId, invocationSource: "assignment", triggerDetail: "system" }).returning();
|
||||
const [task] = await db.insert(issues).values({ companyId: f.company.id, title: "Feedback retry",
|
||||
assigneeAgentId: f.agent.id, executionRunId: run!.id, checkoutRunId: run!.id, status: "in_progress" }).returning();
|
||||
await db.update(heartbeatRuns).set({ nativeIssueId: task!.id }).where(eq(heartbeatRuns.id, run!.id));
|
||||
await db.insert(nativeRunFinalizations).values({ runId: run!.id, companyId: f.company.id, issueId: task!.id,
|
||||
phase: "observed", controllerGeneration: 1, leaseExpiresAt: new Date(Date.now() + 60_000) });
|
||||
const [work] = await db.insert(dotRunnerAssignments).values({ companyId: f.company.id, agentId: f.agent.id,
|
||||
bindingId: f.pairing.bindingId, bindingGeneration: 1, runId: run!.id, normalizedSessionId: sessionId, turnId: randomUUID(),
|
||||
controllerGeneration: 1, catalogDigest: "test", status: "accepted", projection: {},
|
||||
acceptBy: new Date(Date.now() + 60_000), expiresAt: new Date(Date.now() + 60_000) }).returning();
|
||||
it.each(["follow_up", "operation_result"] as const)("does not start another routine for %s during accepted work", async kind => {
|
||||
const f = await fixture(); const { work } = await admittedWork(f, "accepted");
|
||||
await db.update(grokBotRoutines).set({ lastContactAt: new Date(Date.now() - 60_000) }).where(eq(grokBotRoutines.bindingId, f.pairing.bindingId));
|
||||
const sourceEventId = randomUUID(); const requestId = randomUUID();
|
||||
if (kind === "follow_up") await db.insert(dotMailboxItems).values({ companyId: f.company.id,
|
||||
bindingId: f.pairing.bindingId, bindingGeneration: 1, assignmentId: work.id,
|
||||
kind: "follow_up", sourceEventId, references: { consumed: false } });
|
||||
else await db.insert(dotRunnerOperations).values({ companyId: f.company.id, assignmentId: work.id, requestId,
|
||||
digest: "synthetic-progress", command: { action: "tool", input: { name: "report_progress" } }, status: "completed",
|
||||
outcome: { status: "completed", result: { recorded: true } }, pendingReturnedAt: new Date(Date.now() - 30_000) });
|
||||
f.delivery.mockClear(); const dispatcher = createGrokRoutineDispatcher(db, () => true, f.delivery);
|
||||
await dispatcher.tick(); await dispatcher.tick();
|
||||
expect(f.delivery).not.toHaveBeenCalled();
|
||||
if (kind === "follow_up") {
|
||||
expect((await dotRunnerBroker(db).mailbox(f.principal)).items).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ sourceEventId, references: { consumed: false } }),
|
||||
]));
|
||||
} else {
|
||||
expect(await dotRunnerBroker(db).operationStatus(f.principal, work.id, requestId)).toEqual({ status: "completed", result: { recorded: true } });
|
||||
expect((await db.select().from(dotRunnerOperations).where(eq(dotRunnerOperations.requestId, requestId)))[0]?.observedAt).not.toBeNull();
|
||||
}
|
||||
expect((await dotRunnerBroker(db).read(f.principal, work.id)).toolUsageInstructions).toContain("paperclip_grok_operation_status");
|
||||
});
|
||||
|
||||
it("cancels a queued wake when its assignment was already accepted", async () => {
|
||||
const f = await fixture(); const { work } = await admittedWork(f, "offered");
|
||||
const queued = await f.service.reserve({ companyId: f.company.id, agentId: f.agent.id,
|
||||
bindingId: f.pairing.bindingId, assignmentId: work.id, kind: "mailbox_updated", sourceEventId: randomUUID() });
|
||||
await db.update(dotRunnerAssignments).set({ status: "accepted", acceptedAt: new Date() }).where(eq(dotRunnerAssignments.id, work.id));
|
||||
f.delivery.mockClear(); await createGrokRoutineDispatcher(db, () => true, f.delivery).tick();
|
||||
expect(f.delivery).not.toHaveBeenCalled();
|
||||
expect((await db.select().from(grokBotDispatches).where(eq(grokBotDispatches.id, queued.id)))[0]?.status).toBe("cancelled");
|
||||
});
|
||||
|
||||
it("cancels an unsent wake when acceptance commits during secret lookup", async () => {
|
||||
const f = await fixture(); const { work } = await admittedWork(f, "offered");
|
||||
const queued = await f.service.reserve({ companyId: f.company.id, agentId: f.agent.id,
|
||||
bindingId: f.pairing.bindingId, assignmentId: work.id, kind: "mailbox_updated", sourceEventId: randomUUID() });
|
||||
const accept = await acceptance(f, work);
|
||||
const original = secretModule.secretService;
|
||||
vi.spyOn(secretModule, "secretService").mockImplementation(database => {
|
||||
const service = original(database);
|
||||
return { ...service, resolveSecretValue: async (...args) => { await accept(); return service.resolveSecretValue(...args); } };
|
||||
});
|
||||
f.delivery.mockClear();
|
||||
await f.service.send(f.company.id, f.agent.id, queued.id);
|
||||
expect(f.delivery).not.toHaveBeenCalled();
|
||||
expect((await db.select().from(grokBotDispatches).where(eq(grokBotDispatches.id, queued.id)))[0]).toMatchObject({ status: "cancelled", httpStatus: null });
|
||||
expect((await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.id, work.id)))[0]?.status).toBe("accepted");
|
||||
expect(await f.service.recoveryNeeded(f.company.id, f.agent.id)).toBe(false);
|
||||
});
|
||||
|
||||
it("serializes acceptance during the final send check without blocking an authenticated reply", async () => {
|
||||
const f = await fixture(); const { work } = await admittedWork(f, "offered");
|
||||
const queued = await f.service.reserve({ companyId: f.company.id, agentId: f.agent.id,
|
||||
bindingId: f.pairing.bindingId, assignmentId: work.id, kind: "mailbox_updated", sourceEventId: randomUUID() });
|
||||
const accept = await acceptance(f, work);
|
||||
let accepting: Promise<void> | undefined; let authorityChecks = 0;
|
||||
const original = externalAuthority.authorizeExternalBotBinding;
|
||||
vi.spyOn(externalAuthority, "authorizeExternalBotBinding").mockImplementation(async (...args) => {
|
||||
const result = await original(...args);
|
||||
if (++authorityChecks === 3) {
|
||||
accepting = accept();
|
||||
await vi.waitFor(async () => {
|
||||
const waiting = await db.execute(sql`select pid from pg_locks where locktype = 'advisory' and not granted`);
|
||||
expect(waiting.length).toBeGreaterThan(0);
|
||||
});
|
||||
}
|
||||
return result;
|
||||
});
|
||||
f.delivery.mockClear();
|
||||
f.delivery.mockImplementation(async () => {
|
||||
expect((await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.id, work.id)))[0]?.status).toBe("offered");
|
||||
await f.service.acknowledge(f.principal, { correlationId: queued.id, bindingId: f.pairing.bindingId, bindingGeneration: 1 });
|
||||
return { delivery: "accepted", httpStatus: 200 };
|
||||
});
|
||||
await f.service.send(f.company.id, f.agent.id, queued.id);
|
||||
expect(accepting).toBeDefined(); await accepting;
|
||||
expect(f.delivery).toHaveBeenCalledTimes(1);
|
||||
expect((await db.select().from(grokBotDispatches).where(eq(grokBotDispatches.id, queued.id)))[0]?.acknowledgedAt).not.toBeNull();
|
||||
expect((await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.id, work.id)))[0]?.status).toBe("accepted");
|
||||
});
|
||||
|
||||
it("allows Stop to fence work while webhook delivery is pending", async () => {
|
||||
const f = await fixture(); const { work } = await admittedWork(f, "offered");
|
||||
const queued = await f.service.reserve({ companyId: f.company.id, agentId: f.agent.id,
|
||||
bindingId: f.pairing.bindingId, assignmentId: work.id, kind: "mailbox_updated", sourceEventId: randomUUID() });
|
||||
const cancelRun = vi.fn().mockResolvedValue(undefined);
|
||||
const broker = dotRunnerBroker(db, { heartbeat: { wakeup: vi.fn(), cancelRun } });
|
||||
f.delivery.mockClear();
|
||||
f.delivery.mockImplementation(async () => {
|
||||
expect(await broker.stopGrokWork(f.company.id, f.agent.id, f.principal.grant.userId, work.runId))
|
||||
.toMatchObject({ stopped: true, remoteTerminationConfirmed: false });
|
||||
expect((await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.id, work.id)))[0]?.status).toBe("fenced");
|
||||
return { delivery: "unknown", httpStatus: null };
|
||||
});
|
||||
await f.service.send(f.company.id, f.agent.id, queued.id);
|
||||
expect(f.delivery).toHaveBeenCalledTimes(1); expect(cancelRun).toHaveBeenCalledTimes(1);
|
||||
expect((await db.select().from(grokBotRoutines).where(eq(grokBotRoutines.bindingId, f.pairing.bindingId)))[0]?.recoveryReason).toBe("work_session");
|
||||
});
|
||||
|
||||
it("accepts a late acknowledgement after the Bot has claimed its offer", async () => {
|
||||
const f = await fixture(); const { work } = await admittedWork(f, "offered");
|
||||
const sent = await f.service.reserve({ companyId: f.company.id, agentId: f.agent.id,
|
||||
bindingId: f.pairing.bindingId, assignmentId: work.id, kind: "mailbox_updated", sourceEventId: randomUUID() });
|
||||
await f.service.send(f.company.id, f.agent.id, sent.id);
|
||||
await db.update(dotRunnerAssignments).set({ status: "accepted", acceptedAt: new Date() }).where(eq(dotRunnerAssignments.id, work.id));
|
||||
expect(await f.service.acknowledge(f.principal, { correlationId: sent.id, bindingId: f.pairing.bindingId,
|
||||
bindingGeneration: 1 })).toMatchObject({ status: "acknowledged", remoteTerminationConfirmed: false });
|
||||
expect((await db.select().from(dotRunnerAssignments).where(eq(dotRunnerAssignments.id, work.id)))[0]?.status).toBe("accepted");
|
||||
});
|
||||
|
||||
it("retries rejected offer delivery explicitly while retaining unread feedback", async () => {
|
||||
const f = await fixture(); const { work } = await admittedWork(f, "offered");
|
||||
const sourceEventId = randomUUID();
|
||||
await db.insert(dotMailboxItems).values({ companyId: f.company.id, bindingId: f.pairing.bindingId, bindingGeneration: 1,
|
||||
assignmentId: work!.id, kind: "follow_up", sourceEventId, references: { consumed: false } });
|
||||
await f.service.reserve({ companyId: f.company.id, agentId: f.agent.id, bindingId: f.pairing.bindingId,
|
||||
assignmentId: work.id, kind: "mailbox_updated", sourceEventId });
|
||||
f.delivery.mockClear(); f.delivery.mockResolvedValueOnce({ delivery: "rejected", httpStatus: 429 });
|
||||
const dispatcher = createGrokRoutineDispatcher(db, () => true, f.delivery);
|
||||
await dispatcher.tick(); await dispatcher.tick();
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { uploadSchema, type PublicMcpTransfers } from "./public-mcp/file-transfers.js";
|
||||
import { grokRoutineService, grokWakeAcknowledgementSchema } from "./grok-bot/routine.js";
|
||||
import { grokRoutineService, grokWakeAcknowledgementSchema, lockGrokRoutineAcceptance } from "./grok-bot/routine.js";
|
||||
import { authorizeExternalBotBinding, authorizeExternalBotOperator, externalBotEnabled } from "./external-bot-authority.js";
|
||||
import { updateAgentConfigurationInTransaction } from "./agent-configuration-transaction.js";
|
||||
import type { heartbeatService } from "./heartbeat.js";
|
||||
@@ -241,11 +241,15 @@ function createBroker(db: Db) {
|
||||
}
|
||||
return { agentId: binding.agentId, runId: assignment.runId };
|
||||
},
|
||||
async assertWakeAuthority(companyId: string, agentId: string, bindingId: string, assignmentId: string) {
|
||||
async assertWakeAuthority(companyId: string, agentId: string, bindingId: string, assignmentId: string, options: { requireOffered?: boolean } = {}) {
|
||||
const [binding] = await db.select().from(bindings).where(and(eq(bindings.id, bindingId), eq(bindings.companyId, companyId), eq(bindings.agentId, agentId), isNull(bindings.revokedAt)));
|
||||
const [grant] = binding?.grantId ? await db.select().from(mcpOauthGrants).where(eq(mcpOauthGrants.id, binding.grantId)) : [];
|
||||
if (!grant) throw fail("Wake connection authority is unavailable.");
|
||||
return authorizeAssignment({ grant, actor: { type: "agent", companyId, agentId }, company: { id: companyId, name: "", issuePrefix: "", status: "active" } }, assignmentId);
|
||||
const authority = await authorizeAssignment({ grant, actor: { type: "agent", companyId, agentId }, company: { id: companyId, name: "", issuePrefix: "", status: "active" } }, assignmentId);
|
||||
if (options.requireOffered && authority.assignment.status !== "offered") {
|
||||
throw fail("This work was already accepted. Read its feedback and operation receipts without another routine wake.");
|
||||
}
|
||||
return authority;
|
||||
},
|
||||
async authorizeBinding(principal: McpPrincipal, companyId: string, bindingId: string) {
|
||||
const b = await principalBinding(principal, false, true);
|
||||
@@ -271,10 +275,12 @@ function createBroker(db: Db) {
|
||||
});
|
||||
},
|
||||
async read(principal: McpPrincipal, assignmentId: string) {
|
||||
const { assignment } = await authorizeAssignment(principal, assignmentId);
|
||||
const { binding, assignment } = await authorizeAssignment(principal, assignmentId);
|
||||
return { assignmentId, revision: assignment.revision, status: assignment.status,
|
||||
leaseInstructions: "Renew before expiry using the renewal tool for this provider (paperclip_grok_renew or paperclip_dot_renew) and a stable requestId. Renewal is bounded to 24 hours from admission and cannot revive an expired or fenced run.",
|
||||
expiresAt: assignment.expiresAt, ...assignment.projection, accounting: { usage: null, cost: null } };
|
||||
expiresAt: assignment.expiresAt, ...assignment.projection,
|
||||
...(binding.providerKind === "grok_bot" ? { toolUsageInstructions: GROK_BOT_AGENT_TOOL_GUIDANCE } : {}),
|
||||
accounting: { usage: null, cost: null } };
|
||||
},
|
||||
async operation(principal: McpPrincipal, assignmentId: string, requestId: string, action: ExternalProviderOperation["action"], input: Record<string, unknown>) {
|
||||
const { binding, assignment: a } = await authorizeAssignment(principal, assignmentId);
|
||||
@@ -309,7 +315,7 @@ function createBroker(db: Db) {
|
||||
await new Promise(r => setTimeout(r, 50));
|
||||
}
|
||||
if (binding.providerKind === "grok_bot") await db.update(operations).set({ pendingReturnedAt: new Date() }).where(eq(operations.id, row.id));
|
||||
return { status: "pending", assignmentId, requestId, message: binding.providerKind === "grok_bot" ? "Use paperclip_grok_operation_status or retry with the same requestId. Do not create a new request ID." : "Use paperclip_dot_operation_status or retry with the same requestId. Do not create a new request ID." };
|
||||
return { status: "pending", assignmentId, requestId, message: binding.providerKind === "grok_bot" ? "Poll paperclip_grok_operation_status with this assignmentId and requestId until its receipt is terminal. Keep handling this routine; Paperclip will not start another routine for a tool result. Do not repeat the operation with a new request ID or perform dependent work before its receipt completes." : "Use paperclip_dot_operation_status or retry with the same requestId. Do not create a new request ID." };
|
||||
},
|
||||
async tasks(principal: McpPrincipal, after: string | null = null) {
|
||||
const b = await principalBinding(principal);
|
||||
@@ -660,7 +666,17 @@ function createBroker(db: Db) {
|
||||
settle: async event => {
|
||||
const p = z.object({ requestId: z.uuid(), binding: z.object({ runId: z.literal(runId), bindingId: z.literal(execution.provider.binding.bindingId),
|
||||
bindingGeneration: z.literal(execution.provider.binding.bindingGeneration) }).passthrough(), outcome: z.record(z.string(), z.unknown()) }).parse(event.payload);
|
||||
// Identify an immutable reserved accept command before taking any row
|
||||
// locks. A webhook may hold this advisory lock while awaiting HTTP;
|
||||
// Stop and acknowledgements must remain free to lock the binding.
|
||||
const [grokAccept] = await db.select({ id: operations.id }).from(operations)
|
||||
.innerJoin(assignments, eq(assignments.id, operations.assignmentId))
|
||||
.innerJoin(bindings, eq(bindings.id, assignments.bindingId))
|
||||
.where(and(eq(assignments.runId, runId), eq(bindings.id, p.binding.bindingId),
|
||||
eq(bindings.providerKind, "grok_bot"), eq(operations.requestId, p.requestId),
|
||||
sql`${operations.command}->>'action' = 'accept'`));
|
||||
await db.transaction(async tx => {
|
||||
if (grokAccept) await lockGrokRoutineAcceptance(tx as unknown as Db, p.binding.bindingId);
|
||||
// Binding precedes assignment everywhere: serialize mailbox IDs
|
||||
// with dispatch, readiness and reconnect without reversing locks.
|
||||
await tx.select({ id: bindings.id }).from(bindings).where(eq(bindings.id, p.binding.bindingId)).for("update");
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { and, asc, eq, isNull, isNotNull, lt, sql } from "drizzle-orm";
|
||||
import { grokBotDispatches, grokBotRoutines, dotAgentBindings, dotRunnerAssignments, dotRunnerOperations, dotMailboxItems, heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { and, asc, eq, isNull, lt, sql } from "drizzle-orm";
|
||||
import { grokBotDispatches, grokBotRoutines, dotAgentBindings, dotRunnerAssignments, heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { grokRoutineService } from "./routine.js";
|
||||
import type { deliverGrokRoutineWake } from "./webhook.js";
|
||||
import { externalBotEnabled } from "../external-bot-authority.js";
|
||||
@@ -25,43 +25,9 @@ export function createGrokRoutineDispatcher(db: Db, isBackgroundWorkEnabled: ()
|
||||
try { await service.reserve({ companyId: assignment.companyId, agentId: assignment.agentId, bindingId: assignment.bindingId,
|
||||
assignmentId: assignment.id, kind: "mailbox_updated", sourceEventId: `proactive-admission:${assignment.id}` }); } catch { /* Current authority is rechecked on reservation and send. */ }
|
||||
}
|
||||
// Comments and their deferred wake commit independently of egress.
|
||||
// Discover committed, unread input here; recovery never erases feedback.
|
||||
const followUps = await db.select({ item: dotMailboxItems, assignment: dotRunnerAssignments }).from(dotMailboxItems)
|
||||
.innerJoin(dotRunnerAssignments, and(eq(dotRunnerAssignments.id, dotMailboxItems.assignmentId),
|
||||
eq(dotRunnerAssignments.bindingGeneration, dotMailboxItems.bindingGeneration)))
|
||||
.innerJoin(dotAgentBindings, eq(dotAgentBindings.id, dotRunnerAssignments.bindingId))
|
||||
.where(and(eq(dotAgentBindings.providerKind, "grok_bot"), isNull(dotAgentBindings.revokedAt),
|
||||
eq(dotMailboxItems.kind, "follow_up"), eq(dotRunnerAssignments.status, "accepted"),
|
||||
sql`coalesce(${dotMailboxItems.references}->>'consumed', 'false') <> 'true'`,
|
||||
sql`not exists (select 1 from ${grokBotDispatches} where ${grokBotDispatches.assignmentId} = ${dotRunnerAssignments.id} and ${grokBotDispatches.status} in ('prepared', 'sending', 'accepted', 'unknown'))`,
|
||||
sql`not exists (select 1 from ${grokBotDispatches} where ${grokBotDispatches.bindingId} = ${dotRunnerAssignments.bindingId} and ${grokBotDispatches.sourceEventId} = ${dotMailboxItems.sourceEventId})`))
|
||||
.orderBy(asc(dotMailboxItems.id)).limit(20);
|
||||
const feedbackAssignments = new Set<string>();
|
||||
for (const { item, assignment } of followUps) {
|
||||
if (feedbackAssignments.has(assignment.id)) continue;
|
||||
feedbackAssignments.add(assignment.id);
|
||||
try { await service.reserve({ companyId: assignment.companyId, agentId: assignment.agentId, bindingId: assignment.bindingId,
|
||||
assignmentId: assignment.id, kind: "mailbox_updated", sourceEventId: item.sourceEventId }); } catch { /* The durable comment/wake remains queued behind recovery or current authority. */ }
|
||||
}
|
||||
const results = await db.select({ operation: dotRunnerOperations, assignment: dotRunnerAssignments }).from(dotRunnerOperations)
|
||||
.innerJoin(dotRunnerAssignments, eq(dotRunnerAssignments.id, dotRunnerOperations.assignmentId))
|
||||
.innerJoin(dotAgentBindings, eq(dotAgentBindings.id, dotRunnerAssignments.bindingId))
|
||||
.innerJoin(grokBotRoutines, eq(grokBotRoutines.bindingId, dotAgentBindings.id))
|
||||
.where(and(eq(dotAgentBindings.providerKind, "grok_bot"), isNull(dotAgentBindings.revokedAt), eq(dotRunnerAssignments.status, "accepted"),
|
||||
lt(grokBotRoutines.lastContactAt, new Date(Date.now() - 15_000)),
|
||||
sql`not exists (select 1 from ${grokBotDispatches} where ${grokBotDispatches.assignmentId} = ${dotRunnerAssignments.id} and ${grokBotDispatches.status} in ('prepared', 'sending', 'accepted', 'unknown'))`,
|
||||
sql`not exists (select 1 from ${grokBotDispatches} where ${grokBotDispatches.sourceEventId} = 'operation-result:' || ${dotRunnerOperations.id}::text and ${grokBotDispatches.bindingId} = ${dotRunnerAssignments.bindingId})`,
|
||||
eq(dotRunnerOperations.status, "completed"), isNotNull(dotRunnerOperations.pendingReturnedAt), isNull(dotRunnerOperations.observedAt),
|
||||
lt(dotRunnerOperations.pendingReturnedAt, new Date(Date.now() - 5_000)))).limit(20);
|
||||
const notifiedAssignments = new Set<string>();
|
||||
for (const { operation, assignment } of results) {
|
||||
if (notifiedAssignments.has(assignment.id)) continue;
|
||||
notifiedAssignments.add(assignment.id);
|
||||
if (operation.command.action !== "tool") continue;
|
||||
try { await service.reserve({ companyId: assignment.companyId, agentId: assignment.agentId, bindingId: assignment.bindingId,
|
||||
assignmentId: assignment.id, kind: "mailbox_updated", sourceEventId: `operation-result:${operation.id}` }); } catch { /* Never change a receipt or replay the tool. */ }
|
||||
}
|
||||
// An accepted session already has a routine handling it. Quiet MCP
|
||||
// traffic does not prove that routine ended. Keep feedback and receipts
|
||||
// durable for the Bot to read instead of starting overlapping work.
|
||||
const rows = await db.select({ dispatch: grokBotDispatches, agentId: grokBotRoutines.agentId }).from(grokBotDispatches)
|
||||
.innerJoin(grokBotRoutines, and(eq(grokBotRoutines.id, grokBotDispatches.routineId), eq(grokBotRoutines.companyId, grokBotDispatches.companyId)))
|
||||
.where(eq(grokBotDispatches.status, "prepared")).orderBy(asc(grokBotDispatches.createdAt)).limit(10);
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { createHash, randomBytes, randomUUID, timingSafeEqual } from "node:crypto";
|
||||
import { and, asc, desc, eq, inArray, isNull, isNotNull, lte, or } from "drizzle-orm";
|
||||
import { and, asc, desc, eq, inArray, isNull, isNotNull, lte, or, sql } from "drizzle-orm";
|
||||
import { z } from "zod";
|
||||
import { companies, dotAgentBindings as bindings, dotRunnerAssignments as assignments,
|
||||
grokBotRoutines as routines, grokBotDispatches as dispatches, mcpOauthGrants, type Db } from "@paperclipai/db";
|
||||
@@ -17,6 +17,12 @@ export const grokWakeAcknowledgementSchema = z.object({ correlationId: z.uuid(),
|
||||
type Ack = z.infer<typeof grokWakeAcknowledgementSchema>;
|
||||
type Delivery = typeof deliverGrokRoutineWake;
|
||||
|
||||
/** Serialize work acceptance with the bounded HTTP handoff, without holding
|
||||
* binding/assignment row locks that would block Stop or authenticated replies. */
|
||||
export async function lockGrokRoutineAcceptance(database: Db, bindingId: string) {
|
||||
await database.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`grok-routine-acceptance:${bindingId}`}, 0))`);
|
||||
}
|
||||
|
||||
/** Existing shared agent grants and assignments own authority; routines only wake the Bot. */
|
||||
export function grokRoutineService(db: Db, deliver: Delivery = deliverGrokRoutineWake) {
|
||||
async function current(database: Db, companyId: string, agentId: string, bindingId: string, lock = false, allowPaused = false) {
|
||||
@@ -165,7 +171,7 @@ export function grokRoutineService(db: Db, deliver: Delivery = deliverGrokRoutin
|
||||
if (row.kind === "mailbox_updated") {
|
||||
if (!row.assignmentId) throw fail("Work wake requires an admitted assignment.");
|
||||
const { dotRunnerBroker } = await import("../dot-runner-broker.js");
|
||||
await dotRunnerBroker(db).assertWakeAuthority(companyId, agentId, row.bindingId, row.assignmentId);
|
||||
await dotRunnerBroker(db).assertWakeAuthority(companyId, agentId, row.bindingId, row.assignmentId, { requireOffered: true });
|
||||
}
|
||||
// Accepted, acknowledged and uncertain work is never sent again after restart.
|
||||
if (row.expiresAt <= new Date()) throw fail("Unsent dispatch expired. Inspect before creating a new verification.");
|
||||
@@ -178,18 +184,32 @@ export function grokRoutineService(db: Db, deliver: Delivery = deliverGrokRoutin
|
||||
return row ? state(companyId, agentId, row.bindingId) : null;
|
||||
}
|
||||
let networkAttempted = false;
|
||||
let outcome: GrokRoutineDelivery = { delivery: "unknown", httpStatus: null };
|
||||
let outcome: { delivery: GrokRoutineDelivery["delivery"] | "cancelled"; httpStatus: number | null } = { delivery: "unknown", httpStatus: null };
|
||||
try {
|
||||
const secret = await secretService(db).resolveSecretValue(companyId, intent.routine.secretId, intent.routine.secretVersion,
|
||||
{ accessContext: { consumerType: "agent", consumerId: agentId, configPath: "grok.routine", actorType: "system", actorId: "grok-dispatch" } });
|
||||
// Recheck the grant after resolving credentials; resolution is not authority.
|
||||
await current(db, companyId, agentId, intent.row.bindingId);
|
||||
if (intent.row.kind === "mailbox_updated") {
|
||||
const { dotRunnerBroker } = await import("../dot-runner-broker.js");
|
||||
await dotRunnerBroker(db).assertWakeAuthority(companyId, agentId, intent.row.bindingId, intent.row.assignmentId!);
|
||||
}
|
||||
networkAttempted = true;
|
||||
outcome = await deliver(JSON.parse(secret), intent.row.payload as GrokRoutineWake);
|
||||
const handoff = async (database: Db) => {
|
||||
// Acceptance may have committed while credentials were resolving. An
|
||||
// obsolete, unsent wake is cancellation, not a provider rejection.
|
||||
if (intent.row.kind === "mailbox_updated") {
|
||||
const [assignment] = await database.select({ status: assignments.status }).from(assignments)
|
||||
.where(and(eq(assignments.id, intent.row.assignmentId!), eq(assignments.bindingId, intent.row.bindingId)));
|
||||
if (assignment && assignment.status !== "offered") return { delivery: "cancelled" as const, httpStatus: null };
|
||||
}
|
||||
// Recheck after acquiring the acceptance lock and resolving credentials.
|
||||
await current(database, companyId, agentId, intent.row.bindingId);
|
||||
if (intent.row.kind === "mailbox_updated") {
|
||||
const { dotRunnerBroker } = await import("../dot-runner-broker.js");
|
||||
await dotRunnerBroker(database).assertWakeAuthority(companyId, agentId, intent.row.bindingId, intent.row.assignmentId!, { requireOffered: true });
|
||||
}
|
||||
networkAttempted = true;
|
||||
return deliver(JSON.parse(secret), intent.row.payload as GrokRoutineWake);
|
||||
};
|
||||
outcome = intent.row.kind === "mailbox_updated" ? await db.transaction(async tx => {
|
||||
const database = tx as unknown as Db;
|
||||
await lockGrokRoutineAcceptance(database, intent.row.bindingId);
|
||||
return handoff(database);
|
||||
}) : await handoff(db);
|
||||
} catch { outcome = { delivery: networkAttempted ? "unknown" : "rejected", httpStatus: null }; /* Do not persist exception or response text. */ }
|
||||
await db.transaction(async tx => {
|
||||
await tx.select({ id: bindings.id }).from(bindings).where(eq(bindings.id, intent.row.bindingId)).for("update");
|
||||
@@ -215,7 +235,7 @@ export function grokRoutineService(db: Db, deliver: Delivery = deliverGrokRoutin
|
||||
if (row.dispatch.kind === "mailbox_updated") {
|
||||
if (!row.dispatch.assignmentId) throw fail("Work wake requires an admitted assignment.");
|
||||
const { dotRunnerBroker } = await import("../dot-runner-broker.js");
|
||||
await dotRunnerBroker(db).assertWakeAuthority(companyId, agentId, binding.id, row.dispatch.assignmentId);
|
||||
await dotRunnerBroker(db).assertWakeAuthority(companyId, agentId, binding.id, row.dispatch.assignmentId, { requireOffered: true });
|
||||
}
|
||||
// Retain the rejected receipt. Concurrent clicks reserve the same next intent;
|
||||
// accepted/unknown work can never enter this path, and is never replayed.
|
||||
|
||||
@@ -10,6 +10,8 @@ describe("Grok Bot onboarding instructions", () => {
|
||||
expect(routine).toContain("paperclip_grok_ack_wake");
|
||||
expect(routine).toContain("bindingGeneration");
|
||||
expect(routine).toContain("paperclip_grok_request_turn");
|
||||
expect(routine).toContain("paperclip_grok_operation_status");
|
||||
expect(routine).toContain("before finishing, drain paperclip_grok_inbox");
|
||||
expect(routine).not.toContain("paperclip_dot_");
|
||||
expect(routine).not.toContain("Bearer ");
|
||||
});
|
||||
|
||||
Reference in new issue
Block a user