From e9ab3fae619d9abb657096f535fdfdc0c91ffc75 Mon Sep 17 00:00:00 2001 From: Jonathan Date: Tue, 13 Oct 2026 19:37:09 -0400 Subject: [PATCH] fix: deduplicate recovery and protect shared Git workspaces --- CHANGELOG.md | 2 + docs/api.md | 4 +- docs/operations.md | 18 ++- docs/skills.md | 1 + schema/skillhook.yaml.schema.json | 3 + skills/skillhook-authoring/SKILL.md | 9 ++ src/cloud/control.ts | 7 +- src/jobs.ts | 2 +- src/manual.ts | 3 +- src/mcp.ts | 2 +- src/queue.ts | 60 ++++++-- src/replay.ts | 19 ++- src/runtime-safety.test.ts | 207 ++++++++++++++++++++++++++++ src/server.test.ts | 9 +- src/server.ts | 9 +- src/skills.test.ts | 10 ++ src/skills.ts | 2 + 17 files changed, 331 insertions(+), 36 deletions(-) create mode 100644 src/runtime-safety.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 3768f80..0763af6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ All notable changes to skillhook, newest first. The format follows [Keep a Chang ## Unreleased +- Reuse identical in-flight replay work in HTTP and cloud recovery; alternate eligible fresh/replay queue slots and serialize shared Git worktrees by default, with `serialize_cwd` overrides for coordinated parallel jobs or non-Git directories. + ## 0.10.1 (2026-10-13) - Fallback targets accept a runner, model and effort. Recognise Claude session/weekly limits and exhausted credits, ignore synthetic API error notices when checking whether fallback is safe, and never revisit a failed runner in the fallback chain. diff --git a/docs/api.md b/docs/api.md index 840096e..be1f583 100644 --- a/docs/api.md +++ b/docs/api.md @@ -416,7 +416,7 @@ The log lives in `jobs/.delivery-log/` and keeps the newest `deliveries.max` (20 ## `POST /deliveries//replay` -Runs a recorded delivery again through the skill as it is now: the original payload, headers (redacted, plus `x-skillhook-replay-of: `), query string and sender IP, as a new job with `trigger: "replay"`, `source.method: "REPLAY"` and `replay_of: {"delivery": "", "job": ""}` (the job is present when the delivery had been accepted; its `event.json` and `body.bin` are then what is replayed). The signature is not checked again, `when` filters apply unless skipped, and nothing is de-duplicated: a replay never counts as a duplicate and is never folded into a job still in flight, and it carries no `delivery_id` itself. +Runs a recorded delivery again through the skill as it is now: the original payload, headers (redacted, plus `x-skillhook-replay-of: `), query string and sender IP, as a new job with `trigger: "replay"`, `source.method: "REPLAY"` and `replay_of: {"delivery": "", "job": ""}` (the job is present when the delivery had been accepted; its `event.json` and `body.bin` are then what is replayed). The signature is not checked again, `when` filters apply unless skipped, and identical in-flight work is reused when deduplication is enabled. A newly created replay carries no `delivery_id` itself. Body: a JSON object, all fields optional. @@ -427,7 +427,7 @@ Body: a JSON object, all fields optional. | `runner`, `model`, `effort` | string | Overrides for this run, as in `POST /skills//run`. | | `wait` | number | Seconds to wait for the result (also `?wait=`); clamped to `max_wait_seconds`. | -Responses are the webhook shapes (`202` queued, `200` finished when waiting) plus `replay_of`. Errors: `404 unknown_delivery`, `404 unknown_skill` (the skill is gone or disabled), `409 replay_needs_force`, `409 no_body` (the body was not kept: `deliveries.store_bodies` was off, it was cut at `deliveries.body_max_bytes`, or the record was compacted away), `400 bad_request` (not a JSON object, or an unknown `runner`). +Responses are the webhook shapes (`202` queued, `200` finished when waiting) plus `replay_of`. With in-flight deduplication enabled, an identical payload and query in this skill returns the existing job with `duplicate: true, in_flight: true` instead of enqueueing another replay. This applies to HTTP and cloud recovery, including replay chains. Distinct message/source identifiers in the payload stay separate; conversation identity alone is never the key. Overrides and `force` do not bypass deduplication. Finished jobs can be replayed again; the application must still check side-effect receipts. Errors: `404 unknown_delivery`, `404 unknown_skill` (the skill is gone or disabled), `409 replay_needs_force`, `409 no_body` (the body was not kept: `deliveries.store_bodies` was off, it was cut at `deliveries.body_max_bytes`, or the record was compacted away), `400 bad_request` (not a JSON object, or an unknown `runner`). ```bash curl -sS -X POST -H "Authorization: Bearer $SKILLHOOK_ADMIN_TOKEN" -H "Content-Type: application/json" \ diff --git a/docs/operations.md b/docs/operations.md index c69f42d..1235fab 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -179,7 +179,7 @@ skillhook deliveries replay [--force] [--skip-filters] [--runner R] [--mode When a sender reports failures, `skillhook deliveries list --outcome rejected` shows what arrived and why it was refused; `--json` gives the records, `GET /deliveries` the same over the admin API ([api.md](api.md#get-deliveries)), and the MCP tools `list_deliveries` / `get_delivery` the same to an agent. The running server also publishes each record as a `delivery.received` event. -Once the cause is fixed (a secret pasted, a filter corrected, a skill installed), `deliveries replay ` runs the recorded request again through the skill as it is now: a new job with `trigger: replay` and `replay_of`, the original payload, headers and query, no signature check (`--force` for a delivery that was rejected, since its body was never verified), `when` filters applied unless `--skip-filters`, never de-duplicated. `jobs replay ` does the same for any earlier job. Both go through the running server when there is one (`POST /deliveries//replay`, `POST /jobs//replay`; MCP `replay_delivery`, `replay_job`) and run in the CLI process otherwise. The agent is told it is replaying, so a well-written skill checks what earlier runs already did before repeating side effects. On a machine paired with [Skillhook Cloud](https://skillhook.dev) the records go to the cloud as they are written (`cloud.upload_payloads` decides whether the bodies do), where the deliveries of every machine are kept on record together and replayed from the dashboard, the cloud's API, its MCP server or `skillhook cloud replay_delivery `, on the machine they arrived at, on another machine or through another skill: https://skillhook.dev/docs/deliveries. +Once the cause is fixed (a secret pasted, a filter corrected, a skill installed), `deliveries replay ` runs the recorded request again through the skill as it is now: a new job with `trigger: replay` and `replay_of`, the original payload, headers and query, no signature check (`--force` for a delivery that was rejected, since its body was never verified), `when` filters applied unless `--skip-filters`, folded into an identical queued/running job when in-flight deduplication is enabled. Runner overrides and force do not bypass that check. Once that job finishes, another replay may run; this is not a side-effect receipt. `jobs replay ` does the same for any earlier job. Both go through the running server when there is one (`POST /deliveries//replay`, `POST /jobs//replay`; MCP `replay_delivery`, `replay_job`) and run in the CLI process otherwise. The agent is told it is replaying, so a well-written skill checks what earlier runs already did before repeating side effects. On a machine paired with [Skillhook Cloud](https://skillhook.dev) the records go to the cloud as they are written (`cloud.upload_payloads` decides whether the bodies do), where the deliveries of every machine are kept on record together and replayed from the dashboard, the cloud's API, its MCP server or `skillhook cloud replay_delivery `, on the machine they arrived at, on another machine or through another skill: https://skillhook.dev/docs/deliveries. ## Configuration @@ -416,3 +416,19 @@ The agent asked a question (`skillhook jobs show ` prints it) or finished wi The MCP process reads `SKILLHOOK_HOME` or `--dir` from its own configuration. `skillhook mcp --print-config` adds `--dir` when either is set; make the MCP client and your shell agree. Fallback runners also accept targets such as `{ runner: codex, model: gpt-6.1-sol, effort: medium }` (JSON uses quoted keys and values). A target overrides the original model and effort when selected; plain runner names keep the original settings. + +### Recovery queue fairness and shared working directories + +The server alternates eligible fresh work and replays when both are waiting, +keeping FIFO within each class. One replay may precede a fresh arrival, but a +recovery batch cannot monopolize subsequent slots; a continuous fresh stream +cannot starve replays. Running jobs are never preempted. Skill concurrency, +exclusive cwd ownership and the global cap determine eligibility first. + +Git worktrees are serialized by default, including jobs in subdirectories of the same worktree; independent worktrees still run concurrently. `serialize_cwd: false` opts out for jobs that coordinate their edits. Set `serialize_cwd: true` for shared mutable directories outside Git. +The server reserves the canonical worktree root (or resolved cwd outside Git) before asynchronous runner +preparation and holds it through retries, cancellation and terminal completion. +An exclusive job waits for any job in the same cwd, and blocks other jobs there, +including other skills; unrelated directories still run concurrently. The guard +covers one server queue, not direct local processes, multiple servers or external cleanup. Outside Git, nested +working directories are separate keys. It does not recover previously lost edits. diff --git a/docs/skills.md b/docs/skills.md index fcaa209..51a41b5 100644 --- a/docs/skills.md +++ b/docs/skills.md @@ -75,6 +75,7 @@ Unknown top-level keys are allowed. Unknown keys inside `skillhook:` are rejecte | `when` | list of conditions | none | All conditions must hold or the delivery is acknowledged with `200 {"skipped": true}`. See [Filters](#filters-when). | | `env` | list of env var names | `[]` | Variables from `.env` (or the server's environment) exposed to the agent. Nothing else secret is forwarded. See [Environment](#what-the-agent-receives). | | `concurrency` | integer >= 1 | 1 | How many jobs of this skill may run at once. The global cap is `concurrency` in `skillhook.json` (default 2). | +| `serialize_cwd` | boolean | true for Git worktrees, false otherwise | Reserve the resolved working directory exclusively in the server queue, even across skills and symlink aliases. Git jobs reserve their worktree root, including jobs in subdirectories; separate worktrees remain independent. Set false only when parallel jobs coordinate their edits. It takes precedence over per-skill concurrency; either job requesting exclusivity blocks overlap. Independent directories can still run concurrently. | | `dedupe` | `{ path?: string, header?: string, in_flight?: boolean }` | provider delivery-id header; `in_flight` from `jobs.dedupe_in_flight` (`true`) | Where to take the delivery id for replay protection, and whether a delivery identical to a job that is still queued or running is folded into that job. See [Deduplication](#deduplication). | | `claude` | object | — | Claude-only options, below. | | `codex` | object | — | Codex-only options, below. | diff --git a/schema/skillhook.yaml.schema.json b/schema/skillhook.yaml.schema.json index 16327af..ab2355e 100644 --- a/schema/skillhook.yaml.schema.json +++ b/schema/skillhook.yaml.schema.json @@ -424,6 +424,9 @@ "minimum": 1, "maximum": 9007199254740991 }, + "serialize_cwd": { + "type": "boolean" + }, "dedupe": { "type": "object", "properties": { diff --git a/skills/skillhook-authoring/SKILL.md b/skills/skillhook-authoring/SKILL.md index b56df38..d668d65 100644 --- a/skills/skillhook-authoring/SKILL.md +++ b/skills/skillhook-authoring/SKILL.md @@ -183,3 +183,12 @@ To iterate on a draft before installing it, or to try a change without touching Keep SKILL.md under about 150 lines; move API shapes, field lists and long procedures into `references/*.md` and tell the agent when to read them. Fallback runners also accept targets such as `{ runner: codex, model: gpt-6.1-sol, effort: medium }` (JSON uses quoted keys and values). A target overrides the original model and effort when selected; plain runner names keep the original settings. + +Git worktrees are exclusive by default, including jobs below the worktree root. +For shared mutable directories outside Git, set `skillhook.serialize_cwd: true`; +set false only when parallel jobs coordinate their edits. +The server reserves the resolved cwd before preparing the run and holds it through +retries and terminal completion, even across different skills and symlink aliases. +Independent Git worktrees remain parallel. This is one server's queue guard; direct local runs, a second server and external +workspace cleanup still need their own coordination. Keep `concurrency: 1` until +the installed server accepts this field. Validate before deploying a new hook map. diff --git a/src/cloud/control.ts b/src/cloud/control.ts index 228c914..5a13396 100644 --- a/src/cloud/control.ts +++ b/src/cloud/control.ts @@ -16,7 +16,7 @@ import { generateSecretFor, resolveSecretName, setSecret, type Ops } from "../op import type { Paths } from "../paths.js"; import type { JobQueue } from "../queue.js"; import type { SkillRegistry } from "../registry.js"; -import { planReplay, ReplayError } from "../replay.js"; +import { enqueueReplay, planReplay, ReplayError } from "../replay.js"; import type { RunOverrides } from "../run.js"; import { buildSchedulePayload } from "../scheduler.js"; import type { ServerControl } from "../server.js"; @@ -81,9 +81,8 @@ export function createControlHandlers(deps: ControlDeps): Partial replayTool("delivery", input)), ); diff --git a/src/queue.ts b/src/queue.ts index 79f9a8c..4d767fd 100644 --- a/src/queue.ts +++ b/src/queue.ts @@ -1,6 +1,8 @@ import { spawn, type ChildProcess } from "node:child_process"; import { EventEmitter } from "node:events"; -import { createWriteStream, existsSync } from "node:fs"; +import { createWriteStream, existsSync, realpathSync, readFileSync } from "node:fs"; +import path from "node:path"; +import { deliveryFingerprint } from "./payload.js"; import type { Config, RunnerName } from "./config.js"; import type { Secrets } from "./env.js"; import { Events } from "./events.js"; @@ -9,7 +11,7 @@ import type { Logger } from "./logger.js"; import { answerQuestion, readTimeline, type JobAnswer, type JobQuestion, type ProgressEntry } from "./progress.js"; import { deriveOutcome, resolveJobResponse } from "./response.js"; import type { ReadinessCache, RunnerReadiness } from "./readiness.js"; -import { prepareRun } from "./run.js"; +import { prepareRun, resolveRunSettings } from "./run.js"; import { classifyFailure, type FailureKind, type FallbackConfig, type FallbackTrigger, type JobFailure } from "./runners/failure.js"; import type { RunnerOutcome, StreamState } from "./runners/index.js"; import { lastLines } from "./runners/types.js"; @@ -53,6 +55,7 @@ function fallbackPolicy(skill: Skill, config: Config): { runners: FallbackConfig interface Running { job: JobRecord; + workspace?: { cwd: string; key: string; exclusive: boolean }; child?: ChildProcess; cancelled: boolean; timedOut: boolean; @@ -70,13 +73,14 @@ const PROGRESS_POLL_MS = 1000; const WAIT_GRACE_MS = 30_000; /** - * In-memory FIFO with a global concurrency cap and one-at-a-time per skill by default. Jobs are + * In-memory queue with FIFO within fresh/replay classes, bounded fairness, a global concurrency cap and one-at-a-time per skill by default. Jobs are * persisted before they are enqueued, so a crash loses nothing but the running processes. */ export class JobQueue extends EventEmitter { private queued: JobRecord[] = []; private running = new Map(); private stopping = false; + private lastStartedWasReplay = true; private watcher?: NodeJS.Timeout; /** Typed `job.*` events (`job.queued`, `job.started`, `job.updated`, `job.cancelled`, `job.finished`). */ readonly events: Events; @@ -111,8 +115,17 @@ export class JobQueue extends EventEmitter { /** The running (preferred) or queued job of this skill with the same delivery fingerprint, if any. */ findInFlight(skill: string, fingerprint: string): JobRecord | undefined { - for (const running of this.running.values()) if (running.job.skill === skill && running.job.fingerprint === fingerprint) return running.job; - return this.queued.find((job) => job.skill === skill && job.fingerprint === fingerprint); + const matches = (job: JobRecord) => { + if (job.skill !== skill) return false; + if (job.fingerprint) return job.fingerprint === fingerprint; + // Jobs retained across an upgrade may predate manual-job fingerprints. + const event = this.deps.store.readEvent(job.id); + const bodyFile = this.deps.store.pathsFor(job.id).body; + const rawBody = event.body_kind === "binary" && existsSync(bodyFile) ? readFileSync(bodyFile) : undefined; + return deliveryFingerprint({ kind: event.body_kind, payload: event.payload, query: event.query, ...(rawBody ? { rawBody } : {}) }) === fingerprint; + }; + for (const running of this.running.values()) if (matches(running.job)) return running.job; + return this.queued.find(matches); } cancel(id: string): boolean { @@ -200,16 +213,37 @@ export class JobQueue extends EventEmitter { } } + private workspaceFor(job: JobRecord): Running["workspace"] { + try { + const skill = job.adhoc ? loadAdhocSkill(this.deps.store.pathsFor(job.id).skillDir, job.id, job.skill) : this.deps.registry.get(job.skill); + if (!skill) return undefined; // execute records the missing/invalid skill failure + const cwd = path.resolve(resolveRunSettings(skill, this.deps.config, { cwd: job.cwd }).cwd); + const canonical = existsSync(cwd) ? realpathSync(cwd) : cwd; + // Worktree roots, not shared git object stores: sibling worktrees can run + // independently, but a job in backend/ must not race cleanup at the root. + let root = canonical; + while (!existsSync(path.join(root, ".git")) && path.dirname(root) !== root) root = path.dirname(root); + const gitRoot = existsSync(path.join(root, ".git")) ? root : undefined; + return { cwd: canonical, key: gitRoot ?? canonical, exclusive: skill.config.serialize_cwd ?? Boolean(gitRoot) }; + } catch { return undefined; } + } + private tick(): void { if (this.stopping) return; while (this.running.size < this.deps.config.concurrency) { - const index = this.queued.findIndex((job) => { + const eligible = this.queued.map((job, index) => ({ job, index, workspace: this.workspaceFor(job) })).filter(({ job, workspace }) => { const runningForSkill = [...this.running.values()].filter((r) => r.job.skill === job.skill).length; - return runningForSkill < this.perSkillLimit(job.skill); + if (runningForSkill >= this.perSkillLimit(job.skill)) return false; + return !workspace || ![...this.running.values()].some((r) => r.workspace?.key === workspace.key && (workspace.exclusive || r.workspace.exclusive)); }); - if (index < 0) break; - const [job] = this.queued.splice(index, 1); - if (job) void this.execute(job); + // Alternate when both classes can run: a batch cannot starve fresh work, + // and a continuous stream of fresh work cannot starve recovery either. + const preferred = eligible.find(({ job }) => (job.trigger === "replay") !== this.lastStartedWasReplay); + const next = preferred ?? eligible[0]; + if (!next) break; + this.queued.splice(next.index, 1); + this.lastStartedWasReplay = next.job.trigger === "replay"; + void this.execute(next.job, next.workspace); } } @@ -322,9 +356,9 @@ export class JobQueue extends EventEmitter { while (Date.now() < until && !running.cancelled && !this.stopping) await new Promise((r) => setTimeout(r, Math.min(100, until - Date.now()))); } - private async execute(job: JobRecord): Promise { + private async execute(job: JobRecord, workspace?: Running["workspace"]): Promise { const { store, config, registry, logger } = this.deps; - const running: Running = { job, cancelled: false, timedOut: false, progressOffset: 0 }; + const running: Running = { job: workspace ? { ...job, cwd: workspace.cwd } : job, workspace, cancelled: false, timedOut: false, progressOffset: 0 }; this.running.set(job.id, running); this.ensureWatcher(); @@ -415,7 +449,7 @@ export class JobQueue extends EventEmitter { let prepared: ReturnType; try { - prepared = prepareRun({ skill, config, secrets: this.deps.secrets(), fileSecrets: this.deps.fileSecrets?.(), store, job, event: store.readEvent(job.id), cwd: job.cwd, processEnv: this.deps.processEnv }); + prepared = prepareRun({ skill, config, secrets: this.deps.secrets(), fileSecrets: this.deps.fileSecrets?.(), store, job, event: store.readEvent(job.id), cwd: running.workspace?.cwd ?? job.cwd, processEnv: this.deps.processEnv }); } catch (error) { return fail(errorMessage(error)); } diff --git a/src/replay.ts b/src/replay.ts index 43ac93d..c5e291c 100644 --- a/src/replay.ts +++ b/src/replay.ts @@ -1,14 +1,15 @@ // Replaying what the server already received: a delivery from the delivery log (accepted or not) or an earlier job, // through the skill as it is now. Signatures are not checked again (they were, or the delivery was rejected and needs -// `force`), `when` filters apply unless skipped, and nothing is de-duplicated: a replay is a new job with +// `force`), `when` filters apply unless skipped, and identical in-flight work is reused. Otherwise a replay is a new job with // `trigger: replay` and `replay_of` pointing at the original. import { existsSync, readFileSync } from "node:fs"; import type { Config } from "./config.js"; import type { DeliveryLog, DeliveryRecord } from "./delivery-log.js"; import { describeCondition, evaluateConditions } from "./filters.js"; import type { JobRecord, JobStore } from "./jobs.js"; -import type { ManualRunInput } from "./manual.js"; -import { parseBody, type BodyKind } from "./payload.js"; +import { createManualJob, type ManualRunInput } from "./manual.js"; +import { deliveryFingerprint, parseBody, type BodyKind } from "./payload.js"; +import type { JobQueue } from "./queue.js"; import type { SkillRegistry } from "./registry.js"; import type { RunOverrides } from "./run.js"; import type { Skill } from "./skills.js"; @@ -131,3 +132,15 @@ export function planReplay(deps: ReplayDeps, input: ReplayInput): ReplayPlan { }; return { ok: true, skill, input: runInput, origin }; } + +/** Shared atomic admission for HTTP and cloud replay commands; no await between lookup and enqueue. */ +export function enqueueReplay(deps: { config: Config; store: JobStore; queue: JobQueue }, input: ManualRunInput): { job: JobRecord; duplicate: boolean } { + if (input.skill.config.dedupe?.in_flight ?? deps.config.jobs.dedupe_in_flight) { + const fingerprint = deliveryFingerprint({ kind: input.body?.kind ?? (typeof input.payload === "string" ? "text" : "json"), payload: input.payload, rawBody: input.body?.raw, query: input.query }); + const existing = deps.queue.findInFlight(input.skill.name, fingerprint); + if (existing) return { job: deps.store.get(existing.id) ?? existing, duplicate: true }; + } + const job = createManualJob(deps, input); + deps.queue.enqueue(job); + return { job, duplicate: false }; +} diff --git a/src/runtime-safety.test.ts b/src/runtime-safety.test.ts new file mode 100644 index 0000000..b3b77b4 --- /dev/null +++ b/src/runtime-safety.test.ts @@ -0,0 +1,207 @@ +import { execFileSync } from "node:child_process"; +import { existsSync, mkdirSync, readFileSync, symlinkSync, writeFileSync } from "node:fs"; +import path from "node:path"; +import { describe, expect, it } from "vitest"; +import { loadConfig } from "./config.js"; +import { JobStore } from "./jobs.js"; +import { silentLogger } from "./logger.js"; +import { createManualJob } from "./manual.js"; +import type { ReadinessCache } from "./readiness.js"; +import { JobQueue } from "./queue.js"; +import { SkillRegistry } from "./registry.js"; +import { enqueueReplay, planReplay } from "./replay.js"; +import { createControlHandlers } from "./cloud/control.js"; +import type { CommandDeps } from "./cloud/commands.js"; +import type { Command } from "./cloud/protocol.js"; +import { tempHome, writeConfigFile, writeSkill } from "./test-support/helpers.js"; + +// Real subprocesses, gated by files in a disposable home; no real jobs or checkouts. +const command = `const fs=require('node:fs'), p=require('node:path'); +const e=JSON.parse(fs.readFileSync(p.join(process.env.SKILLHOOK_JOB_DIR,'event.json'),'utf8')).payload; +if(e.draft) fs.writeFileSync('draft.txt','sibling edit'); +if(e.clean) fs.rmSync(e.cleanPath||'draft.txt',{force:true}); +fs.writeFileSync(p.join(process.env.SKILLHOOK_JOB_DIR,'spawned'),'1'); +const timer=setInterval(()=>{if(!e.gate || fs.existsSync(e.gate)){clearInterval(timer);if(e.draft && fs.readFileSync('draft.txt','utf8')!=='sibling edit')process.exit(2);process.exit(e.fail?1:0);}},5);`; +function harness(concurrency = 1) { + const paths = tempHome("skillhook-runtime-safety-"); + writeConfigFile(paths, { concurrency }); + const config = loadConfig(paths); + const store = new JobStore(paths.jobsDir, { maxJobs: 1000, dedupeWindowSeconds: 60 }); + const registry = new SkillRegistry(paths.skillsDir); + const secrets = () => ({}); + const queue = new JobQueue({ store, registry, config, secrets, logger: silentLogger }); + const deps = { paths, config, store, registry, queue, secrets, fileSecrets: secrets, logger: silentLogger }; + const skill = (name: string, cwd = paths.home, exclusive?: boolean, dedupe = true) => { + writeSkill(paths, name, `description: fixture\nskillhook:\n runner: shell\n concurrency: 4\n${exclusive === undefined ? "" : ` serialize_cwd: ${exclusive}\n`} dedupe: { in_flight: ${dedupe} }\n cwd: ${JSON.stringify(cwd)}\n shell:\n command: ${JSON.stringify([process.execPath, "-e", command])}`); + return registry.get(name)!; + }; + const job = (name: string, payload: unknown, trigger: "api" | "replay" = "api") => createManualJob(deps, { skill: registry.get(name)!, payload, trigger }); + const gate = path.join(paths.home, "release"); + const release = () => writeFileSync(gate, "1"); + const spawned = async (id: string) => { + const until = Date.now() + 10_000; + while (!existsSync(path.join(store.pathsFor(id).dir, "spawned"))) { + if (Date.now() > until) throw new Error(`job did not spawn: ${id}`); + await new Promise((r) => setTimeout(r, 5)); + } + }; + const finish = async (id: string) => { expect((await queue.waitFor(id, 10_000))?.status).toBe("succeeded"); }; + return { ...deps, skill, job, gate, release, spawned, finish }; +} + +describe("recovery admission and fair scheduling", () => { + it("folds repeated recovery, replay chains and cloud recovery of one source, but keeps separate turns", async () => { + const h = harness(); h.skill("reply"); + const old = h.job("reply", { conversation: "same", source: "one", gate: h.gate }); + h.store.update(old.id, { status: "timed_out" }); + const plan = () => { + const p = planReplay(h, { source: "job", id: old.id }); + if (!p.ok) throw new Error(p.reason); + return p.input; + }; + try { + const first = enqueueReplay(h, plan()); + expect(first.duplicate).toBe(false); + await h.spawned(first.job.id); + expect(enqueueReplay(h, plan())).toMatchObject({ duplicate: true, job: { id: first.job.id } }); + const anotherOriginal = h.job("reply", { source: "one", gate: h.gate, conversation: "same" }); + h.store.update(anotherOriginal.id, { status: "failed" }); + const anotherPlan = planReplay(h, { source: "job", id: anotherOriginal.id }); + if (!anotherPlan.ok) throw new Error(anotherPlan.reason); + expect(enqueueReplay(h, anotherPlan.input).job.id).toBe(first.job.id); + const chain = planReplay(h, { source: "job", id: first.job.id }); + if (!chain.ok) throw new Error(chain.reason); + expect(enqueueReplay(h, chain.input).job.id).toBe(first.job.id); + const handlers = createControlHandlers(h); + const cloud = await handlers["job.replay"]!({ id: old.id } as never, { id: "cloud-recovery", type: "job.replay", issued_at: new Date().toISOString() } as Command, {} as CommandDeps); + expect(cloud.result).toMatchObject({ duplicate: true, in_flight: true, job_id: first.job.id }); + const second = enqueueReplay(h, { ...plan(), payload: { conversation: "same", source: "two", gate: h.gate } }); + expect(second.duplicate).toBe(false); + expect(second.job.id).not.toBe(first.job.id); + expect(enqueueReplay(h, { ...plan(), payload: { source: "two", conversation: "same", gate: h.gate } }).job.id).toBe(second.job.id); + + h.release(); + await h.finish(first.job.id); await h.finish(second.job.id); + const later = enqueueReplay(h, plan()); + expect(later.duplicate).toBe(false); // not a permanent side-effect receipt + await h.finish(later.job.id); + } finally { h.release(); await h.queue.shutdown(); } + }); + + it("reuses a legacy queued binary job without a persisted fingerprint", async () => { + const h = harness(); const skill = h.skill("reply"); + const blocker = h.job("reply", { gate: h.gate }); h.queue.enqueue(blocker); await h.spawned(blocker.id); + const bytes = Buffer.from([0, 255, 128, 10]); + const input = { skill, payload: { binary: true, bytes: 4 }, trigger: "replay" as const, body: { kind: "binary" as const, contentType: "application/octet-stream", raw: bytes } }; + const legacy = createManualJob(h, input); + const retained = h.store.update(legacy.id, { fingerprint: undefined }); + h.queue.enqueue(retained); + try { + expect(enqueueReplay(h, input)).toMatchObject({ duplicate: true, job: { id: legacy.id } }); + expect(enqueueReplay(h, { ...input, body: { ...input.body, raw: Buffer.from([0, 255, 127, 10]) } }).duplicate).toBe(false); + } finally { h.release(); await h.queue.shutdown(); } + }); + + it("alternates eligible fresh turns and recovery, preserving each lane's FIFO", async () => { + const h = harness(); h.skill("reply"); + const blocker = h.job("reply", { gate: h.gate }); + const started: string[] = []; + h.queue.events.on("job.started", ({ data }) => started.push(data.job.id)); + h.queue.enqueue(blocker); await h.spawned(blocker.id); + const replay = [1, 2, 3].map((source) => h.job("reply", { source }, "replay")); + replay.forEach((j) => h.queue.enqueue(j)); + const fresh = [4, 5, 6].map((source) => h.job("reply", { source })); + fresh.forEach((j) => h.queue.enqueue(j)); + try { + h.release(); + for (const j of [blocker, ...replay, ...fresh]) await h.finish(j.id); + // At most one replay runs before the fresh turn, even with a batch ahead. + expect(started).toEqual([blocker.id, replay[0]!.id, fresh[0]!.id, replay[1]!.id, fresh[1]!.id, replay[2]!.id, fresh[2]!.id]); + } finally { h.release(); await h.queue.shutdown(); } + }); +}); + +describe("exclusive cwd ownership", () => { + it("prevents sibling cleanup across skills and symlink aliases while other workspaces run", async () => { + const h = harness(4); + const shared = path.join(h.paths.home, "shared"); mkdirSync(shared); + const alias = path.join(h.paths.home, "alias"); symlinkSync(shared, alias, "dir"); + const other = path.join(h.paths.home, "other"); mkdirSync(other); + h.skill("writer", shared, true); h.skill("cleaner", alias); h.skill("unrelated", other); + const writer = h.job("writer", { draft: true, gate: h.gate }); + const cleaner = h.job("cleaner", { clean: true }); + const independent = h.job("unrelated", {}); + try { + h.queue.enqueue(writer); await h.spawned(writer.id); + h.queue.enqueue(cleaner); h.queue.enqueue(independent); + await h.finish(independent.id); + expect(h.store.require(cleaner.id).status).toBe("queued"); + expect(readFileSync(path.join(shared, "draft.txt"), "utf8")).toBe("sibling edit"); + // An eligible replay can run even though the fresh cleaner is blocked. + const recovery = h.job("unrelated", {}, "replay"); h.queue.enqueue(recovery); await h.finish(recovery.id); + h.release(); await h.finish(writer.id); await h.finish(cleaner.id); + expect(existsSync(path.join(shared, "draft.txt"))).toBe(false); + } finally { h.release(); await h.queue.shutdown(); } + }); + + it("defaults to exclusive Git worktree ownership across root and subdirectory jobs", async () => { + const h = harness(4); + const repo = path.join(h.paths.home, "repo"); mkdirSync(repo); + const git = (...args: string[]) => execFileSync("git", args, { cwd: repo, stdio: "pipe" }); + git("init", "--quiet"); git("config", "user.email", "test@example.invalid"); git("config", "user.name", "Test"); + writeFileSync(path.join(repo, "tracked.txt"), "base"); + git("add", "tracked.txt"); git("commit", "--quiet", "-m", "fixture"); + const sub = path.join(repo, "backend"); mkdirSync(sub); + const independent = path.join(h.paths.home, "independent"); + git("worktree", "add", "--quiet", "--detach", independent, "HEAD"); + h.skill("root", repo); h.skill("child", sub); h.skill("another-tree", independent); + const root = h.job("root", { draft: true, gate: h.gate }); + const child = h.job("child", { clean: true, cleanPath: "../draft.txt" }); + const other = h.job("another-tree", {}); + try { + h.queue.enqueue(root); await h.spawned(root.id); + h.queue.enqueue(child); h.queue.enqueue(other); + await h.finish(other.id); + expect(h.store.require(child.id).status).toBe("queued"); + expect(readFileSync(path.join(repo, "draft.txt"), "utf8")).toBe("sibling edit"); + h.release(); await h.finish(root.id); await h.finish(child.id); + } finally { h.release(); await h.queue.shutdown(); } + }); + + it("keeps the reserved cwd through fallback even if defaults change during readiness", async () => { + const h = harness(4); + const before = path.join(h.paths.home, "before"); const after = path.join(h.paths.home, "after"); + mkdirSync(before); mkdirSync(after); h.config.defaults.cwd = before; + writeSkill(h.paths, "fallback", `description: fallback fixture\nskillhook:\n runner: claude\n serialize_cwd: true\n fallback:\n runners: [shell]\n shell:\n command: ${JSON.stringify([process.execPath, "-e", command])}`); + const readiness = { + get: async (runner: "claude" | "shell") => { + h.config.defaults.cwd = after; // simulate hot reload during asynchronous preflight + return { runner, found: true, authenticated: runner === "shell" ? null : false, ready: runner === "shell", detail: "fixture", checked_at: new Date().toISOString() }; + }, + } as unknown as ReadinessCache; + const queue = new JobQueue({ ...h, readiness }); + const job = h.job("fallback", { draft: true }); + try { + queue.enqueue(job); + const done = await queue.waitFor(job.id, 10_000); + expect(done).toMatchObject({ status: "succeeded", runner: "shell", cwd: before }); + expect(readFileSync(path.join(before, "draft.txt"), "utf8")).toBe("sibling edit"); + expect(existsSync(path.join(after, "draft.txt"))).toBe(false); + } finally { await queue.shutdown(); } + }); + + it("holds the cwd when the exclusive job arrives second, and releases after cancellation and failure", async () => { + const h = harness(4); h.skill("plain"); h.skill("exclusive", h.paths.home, true); + const plain = h.job("plain", { gate: h.gate }); + const exclusive = h.job("exclusive", { fail: true }); + const next = h.job("exclusive", {}); + try { + h.queue.enqueue(plain); await h.spawned(plain.id); + h.queue.enqueue(exclusive); h.queue.enqueue(next); + expect(h.queue.cancel(plain.id)).toBe(true); + expect((await h.queue.waitFor(plain.id, 10_000))?.status).toBe("cancelled"); + expect((await h.queue.waitFor(exclusive.id, 10_000))?.status).toBe("failed"); + await h.finish(next.id); + } finally { h.release(); await h.queue.shutdown(); } + }); +}); diff --git a/src/server.test.ts b/src/server.test.ts index 8aab28b..be928da 100644 --- a/src/server.test.ts +++ b/src/server.test.ts @@ -579,7 +579,7 @@ describe("HTTP surface", () => { const replayJob = store.get(String(forced.job_id))!; expect(replayJob).toMatchObject({ trigger: "replay", replay_of: { delivery: skippedId }, source: { method: "REPLAY", ip: "127.0.0.1" } }); expect(replayJob.delivery_id).toBeUndefined(); - expect(replayJob.fingerprint).toBeUndefined(); + expect(replayJob.fingerprint).toMatch(/^[a-f0-9]{64}$/); const event = store.readEvent(replayJob.id); expect(event).toMatchObject({ trigger: "replay", body_kind: "json", payload: { action: "deleted", marker: "rp-skip" } }); expect(event.headers["x-skillhook-replay-of"]).toBe(skippedId); @@ -594,7 +594,7 @@ describe("HTTP surface", () => { const forcedRejected = await json(await fetch(`${base}/deliveries/${rejectedId}/replay`, { method: "POST", headers: auth, body: JSON.stringify({ force: true, wait: 20, model: "sonnet" }) })); expect(forcedRejected).toMatchObject({ status: "succeeded", replay_of: { delivery: rejectedId } }); expect(String(forcedRejected.result)).toContain("model=sonnet"); - // An accepted delivery replays through its job, a job replays directly, and neither is folded into an in-flight twin. + // An accepted delivery replays through its job, a job replays directly, and finished work can be replayed again. const original = await json(await fetch(`${base}/hooks/hello?wait=20`, { method: "POST", body: '{"name":"rp-job"}', headers: { authorization: "Bearer hello-secret", "content-type": "application/json" } })); const viaJob = await json(await fetch(`${base}/jobs/${original.job_id}/replay`, { method: "POST", headers: auth, body: JSON.stringify({ wait: 20 }) })); expect(viaJob).toMatchObject({ status: "succeeded", replay_of: { job: original.job_id } }); @@ -606,8 +606,7 @@ describe("HTTP surface", () => { await waitForJob(String(viaDelivery.job_id)); const twin = await json(await fetch(`${base}/hooks/twin`, { method: "POST", body: '{"replay":"twin"}', headers: { authorization: "Bearer tw", "content-type": "application/json" } })); const twinReplay = await json(await fetch(`${base}/jobs/${twin.job_id}/replay`, { method: "POST", headers: auth, body: "{}" })); - expect(twinReplay.status).toBe("queued"); - expect(twinReplay.job_id).not.toBe(twin.job_id); + expect(twinReplay).toMatchObject({ duplicate: true, in_flight: true, job_id: twin.job_id }); await waitForJob(String(twin.job_id), 25_000); await waitForJob(String(twinReplay.job_id), 25_000); expect((await fetch(`${base}/jobs/20200101T000000Z-aaaaaa/replay`, { method: "POST", headers: auth, body: "{}" })).status).toBe(404); @@ -615,7 +614,7 @@ describe("HTTP surface", () => { expect((await fetch(`${base}/jobs/${original.job_id}/replay`, { method: "POST", headers: auth, body: JSON.stringify({ runner: "gemini" }) })).status).toBe(400); expect((await fetch(`${base}/jobs/${original.job_id}/replay`, { method: "POST", headers: { "x-forwarded-for": "203.0.113.1", "content-type": "application/json" }, body: "{}" })).status).toBe(401); const replays = (await json(await fetch(`${base}/jobs?trigger=replay&limit=20`, { headers: auth }))) as unknown as { jobs: { trigger: string }[] }; - expect(replays.jobs.length).toBeGreaterThanOrEqual(5); + expect(replays.jobs.length).toBeGreaterThanOrEqual(4); expect(replays.jobs.every((j) => j.trigger === "replay")).toBe(true); }); diff --git a/src/server.ts b/src/server.ts index 3d7b65b..1ba2631 100644 --- a/src/server.ts +++ b/src/server.ts @@ -22,7 +22,7 @@ import { createAdhocJob, createManualJob } from "./manual.js"; import { deliveryFingerprint, parseBody, redactHeaders, TRIGGERS, type BodyKind, type Trigger, type WebhookEvent } from "./payload.js"; import { readProgress } from "./progress.js"; import { collectStats, parseSince } from "./stats.js"; -import { planReplay, ReplayError, replayOfFor, type ReplayPlan } from "./replay.js"; +import { enqueueReplay, planReplay, ReplayError, replayOfFor, type ReplayPlan } from "./replay.js"; import { JOB_OUTCOMES, jobOutcome, type JobOutcome } from "./response.js"; import type { JobQueue } from "./queue.js"; import { resolveRunSettings } from "./run.js"; @@ -577,11 +577,10 @@ export function createServer(deps: ServerDeps): Server { logger.info("replay skipped by filter", { skill: plan.skill.name, replay_of: replayOf, reason: plan.reason }); return send(res, 200, { ok: true, skipped: true, reason: plan.reason, replay_of: replayOf }); } - const job = createManualJob({ config, store }, plan.input); - logger.info("replay accepted", { skill: job.skill, job: job.id, replay_of: replayOf, trigger: job.trigger }); - queue.enqueue(job); + const { job, duplicate } = enqueueReplay({ config, store, queue }, plan.input); + logger.info(duplicate ? "replay already in flight" : "replay accepted", { skill: job.skill, job: job.id, replay_of: replayOf, trigger: job.trigger }); const wait = Math.min(Number(body.wait ?? 0) || parseWait(url, headers, config.max_wait_seconds), config.max_wait_seconds); - return respondWithJob(res, job, wait, { replay_of: replayOf }); + return respondWithJob(res, job, wait, { replay_of: replayOf, ...(duplicate ? { duplicate: true, in_flight: true } : {}) }); } /** `GET /jobs//events`: a `status` snapshot, then `stdout`/`stderr` chunks as the files grow and `status` updates from the bus, then `end`. */ diff --git a/src/skills.test.ts b/src/skills.test.ts index 106e586..1193a77 100644 --- a/src/skills.test.ts +++ b/src/skills.test.ts @@ -156,3 +156,13 @@ describe("schedule options", () => { expect(() => parseSkillDocument(doc(" webhook: false"), "/tmp/s")).toThrow(/needs a `schedule`/); }); }); + +describe("exclusive working directory configuration", () => { + it("accepts explicit serialization without changing the default", () => { + const doc = (value: string) => `---\nname: s\ndescription: s\nskillhook:\n${value}\n---\nBody\n`; + expect(parseSkillDocument(doc(" serialize_cwd: true\n concurrency: 4"), "/tmp/s").config.serialize_cwd).toBe(true); + expect(parseSkillDocument(doc(" serialize_cwd: false"), "/tmp/s").config.serialize_cwd).toBe(false); + expect(parseSkillDocument(doc(" concurrency: 4"), "/tmp/s").config.serialize_cwd).toBeUndefined(); + expect(() => parseSkillDocument(doc(" serialize_cwd: yes"), "/tmp/s")).toThrow(/serialize_cwd/); + }); +}); diff --git a/src/skills.ts b/src/skills.ts index b361432..9d5531a 100644 --- a/src/skills.ts +++ b/src/skills.ts @@ -121,6 +121,8 @@ export const SkillhookBlockSchema = z env: z.array(z.string()).optional(), /** How many jobs of this skill may run at once (default 1). */ concurrency: z.number().int().min(1).optional(), + /** Reserve the working directory exclusively across skills (default true for Git worktrees, false otherwise). */ + serialize_cwd: z.boolean().optional(), dedupe: z .object({ /** Dotted path into the payload whose value identifies the delivery (wins over `header` and the provider header). */