Deduplicate recovery, share queue slots fairly and serialize Git workspaces

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.

docs/api.md

@@ -416,7 +416,7 @@ The log lives in `jobs/.delivery-log/` and keeps the newest `deliveries.max` (20
 
 ## `POST /deliveries/<id>/replay`
 
-Runs a recorded delivery again through the skill as it is now: the original payload, headers (redacted, plus `x-skillhook-replay-of: <delivery id>`), query string and sender IP, as a new job with `trigger: "replay"`, `source.method: "REPLAY"` and `replay_of: {"delivery": "<id>", "job": "<original job id>"}` (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: <delivery id>`), query string and sender IP, as a new job with `trigger: "replay"`, `source.method: "REPLAY"` and `replay_of: {"delivery": "<id>", "job": "<original job id>"}` (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/<skill>/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" \

docs/operations.md

@@ -179,7 +179,7 @@ skillhook deliveries replay <id> [--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 <id>` 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 <id>` does the same for any earlier job. Both go through the running server when there is one (`POST /deliveries/<id>/replay`, `POST /jobs/<id>/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 <id>`, 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 <id>` 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 <id>` does the same for any earlier job. Both go through the running server when there is one (`POST /deliveries/<id>/replay`, `POST /jobs/<id>/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 <id>`, 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 <id>` 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.

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. |

schema/skillhook.yaml.schema.json

@@ -424,6 +424,9 @@
             "minimum": 1,
             "maximum": 9007199254740991
           },
+          "serialize_cwd": {
+            "type": "boolean"
+          },
           "dedupe": {
             "type": "object",
             "properties": {

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.

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<Record<Command
       throw error;
     }
     if (!plan.ok) return { result: { accepted: false, skipped: true, reason: plan.reason } };
-    const job = createManualJob({ config: deps.config, store: deps.store }, { ...plan.input, headers: { ...(plan.input.headers ?? {}), "x-skillhook-cloud-user": requester(command) } });
-    deps.queue.enqueue(job);
-    return { result: { accepted: true, job_id: job.id, skill: job.skill, replay_of: job.replay_of } };
+    const { job, duplicate } = enqueueReplay(deps, { ...plan.input, headers: { ...(plan.input.headers ?? {}), "x-skillhook-cloud-user": requester(command) } });
+    return { result: { accepted: true, job_id: job.id, skill: job.skill, replay_of: job.replay_of, ...(duplicate ? { duplicate: true, in_flight: true } : {}) } };
   };
 
   return {

src/jobs.ts

@@ -80,7 +80,7 @@ export interface JobRecord {
   /** Earlier attempts of this job (a retry, or a fallback after a failed run); the record itself is the last one. */
   attempts?: JobAttempt[];
   delivery_id?: string;
-  /** Hash of payload + query for in-flight de-duplication of webhook deliveries (see `deliveryFingerprint`). */
+  /** Hash of payload + query for in-flight de-duplication of deliveries and recovery (see `deliveryFingerprint`). */
   fingerprint?: string;
   source: JobSource;
 }

src/manual.ts

@@ -6,7 +6,7 @@ import type { Config, RunnerName } from "./config.js";
 import { parseFrontmatter } from "./frontmatter.js";
 import { newJobId } from "./ids.js";
 import type { JobRecord, JobStore } from "./jobs.js";
-import { redactHeaders, type BodyKind, type Trigger, type WebhookEvent } from "./payload.js";
+import { deliveryFingerprint, redactHeaders, type BodyKind, type Trigger, type WebhookEvent } from "./payload.js";
 import type { JobAnswer, JobQuestion } from "./progress.js";
 import { resolveRunSettings } from "./run.js";
 import { parseSkillDocument, SkillError, type Skill } from "./skills.js";
@@ -76,6 +76,7 @@ export function createManualJob(ops: { config: Config; store: JobStore }, input:
     source: { ip: input.sourceIp ?? "127.0.0.1", method: input.sourceMethod ?? "LOCAL", path: event.path, content_type: event.content_type, user_agent: event.headers["user-agent"] },
     delivery_id: input.deliveryId,
     replay_of: input.replayOf,
+    fingerprint: deliveryFingerprint({ kind: event.body_kind, payload: event.payload, rawBody: input.body?.raw, query: event.query }),
     adhoc: input.adhoc,
     skill_file: input.skill.file,
     resume_of: input.resume?.of,

src/mcp.ts

@@ -309,7 +309,7 @@ export function buildMcpServer(paths: Paths, env: NodeJS.ProcessEnv = process.en
 
   server.registerTool(
     "replay_delivery",
-    { title: "Replay delivery", description: "Runs a recorded delivery again through the skill as it is now, as a new job with trigger `replay`: the original payload, headers and query, no signature check, `when` filters applied unless skip_filters, never de-duplicated. A delivery that was rejected needs `force` (its body was never verified). Uses the running server when there is one, otherwise runs in-process.", inputSchema: z.object({ ...replayInput, force: z.boolean().optional().describe("replay a delivery that was rejected or errored") }) },
+    { title: "Replay delivery", description: "Runs a recorded delivery again through the skill as it is now, as a new job with trigger `replay`: the original payload, headers and query, no signature check, `when` filters applied unless skip_filters, identical in-flight work is reused when deduplication is enabled on the running server. A delivery that was rejected needs `force` (its body was never verified). Uses the running server when there is one, otherwise runs in-process.", inputSchema: z.object({ ...replayInput, force: z.boolean().optional().describe("replay a delivery that was rejected or errored") }) },
     wrap((input) => replayTool("delivery", input)),
   );