feat(core): carry the unit's AbortSignal on the ambient record

.changeset/initial-kernel.md

@@ -31,7 +31,7 @@ application scope on every path.
 - `runMain`, which turns an outcome into a process exit code (`0` / `1` / `2` /
   `70`) by setting `process.exitCode`.
 - `currentUnit()` over an `AsyncLocalStorage` record carrying
-  `{ unitId, traceId, tenantId, deadline }` — data, never capabilities.
+  `{ unitId, traceId, tenantId, deadline, signal }` — data, never capabilities.
 - A `@btravstack/testing` package with `testRuntime`,
   `createFakeClock` and `withApp`.
 - **Every async API returns an `AsyncResult`, never a bare `Promise`** — the

.changeset/unit-signal.md

@@ -0,0 +1,37 @@
+---
+"@btravstack/core": minor
+"@btravstack/temporal": patch
+"@btravstack/amqp": patch
+---
+
+**`UnitRecord` gains `signal: AbortSignal`** — the ambient record is five
+fields now, not four. It is the **very** controller the unit's work callback is
+handed, not a copy: one abort, two ways to reach it, fired at the drain
+deadline or at once on a path that skips the drain.
+
+The gap it closes: a middleware-shaped runtime opens its unit around a call it
+does not own the arguments of. `@btravstack/temporal`'s `activityUnits` and
+`@btravstack/amqp`'s `messageUnits` both hand the kernel a work callback that
+_is_ the library's `next()`, so an activity or a handler had no parameter to
+receive the signal through and the kernel's `drainTimeoutMs` was unobservable
+from inside the work. Injecting a context the transport's contract does not
+type was the alternative, and it is exactly the hidden-dependency shape `di`
+exists to prevent, so the signal travels on the record instead — data about
+this unit, like `deadline`, with nothing to substitute in a test.
+`@btravstack/http` is unchanged: it still passes the same signal as its
+handler's third parameter.
+
+What each transport does with it is the transport's own business, and both
+examples are worked:
+
+- **`examples/order-amqp-worker`** answers a `RetryableError` when
+  `currentUnit()?.signal.aborted`, leaving the delivery un-acked so the broker
+  hands it to the next worker. This transport has no cancellation of its own —
+  a redelivery is recovery, not cancellation.
+- **`examples/order-temporal-worker`**'s `ShippingService.arrange` fails as a
+  **defect**, which the platform retries on another worker. The contract's
+  `ShippingUnavailable` is a permanent no and would be the wrong error for "we
+  ran out of time". Temporal's `Context.current().cancellationSignal` is a
+  different clock — workflow-side cancellation, and worker shutdown after
+  `shutdownGraceTime` — so the two are honoured together rather than one
+  standing in for the other.

CLAUDE.md

@@ -93,12 +93,24 @@ hook). User-facing changes need a changeset.
 
 2. **Ambient carries DATA. The DI `Context` carries CAPABILITIES.** The kernel
    opens one `AsyncLocalStorage` store per unit holding a small, fixed record —
-   `{ unitId, traceId, tenantId, deadline }` (`UnitRecord` in `units.ts`) —
+   `{ unitId, traceId, tenantId, deadline, signal }` (`UnitRecord` in
+   `units.ts`) —
    and nothing else. Services never go in it. The line holds because what `di`
    exists to prevent is hidden _dependencies_: code that secretly needs a
    collaborator it never declared and cannot be tested without it. A trace id is
    not a collaborator — no substitutability question, no test double, nothing to
-   swap. A repository pulled from an ambient store is the untestable coupling; a
+   swap. Nor is an `AbortSignal`: `signal` is the **very** controller the work
+   callback is handed — one abort, two ways to reach it — and it is on the
+   record because the callback is not always where the work is. A
+   middleware-shaped runtime (`@btravstack/temporal`, `@btravstack/amqp`) opens
+   the unit around a call it does not own the arguments of, so an activity or a
+   handler has no parameter to receive it through, and injecting a context the
+   contract does not type was the alternative and was rejected;
+   `@btravstack/http` passes the same signal as its handler's third parameter,
+   which is that signal by another route. A transport's own cancellation —
+   Temporal's `Context.current().cancellationSignal` — is a **different clock**,
+   not this one. A repository pulled from an ambient store is the untestable
+   coupling; a
    tenant id read by the Postgres adapter is not. Legitimate readers are
    infrastructure adapters only (logger, OTel exporter, database adapter);
    application code reading the store is meant to be a lint error, in the spirit
@@ -403,6 +415,12 @@ UnitNeeds>`: `NO RUNTIME` when the module exports no runtime port,
   `Result` — the `Promise` arm is Thesis #6's second exception, since it exists
   to accept a caller's `async` handler. `UnitRegistry.awaitIdle()` returns
   `AsyncResult<void, never>`.
+- **`UnitRecord`** — the ambient record: `{ unitId, traceId, tenantId, deadline,
+signal }`. `signal` is the same `AbortSignal` `UnitWork` receives as its
+  argument — aborted at the drain deadline, or at once on a path that skips the
+  drain (`abortAll`) — carried here so a runtime whose work callback is a
+  library's `next()` still reaches it. Guarded by `units.spec.ts` → _"carries
+  the work's own AbortSignal on the ambient record"_.
 - **`currentUnit()` → `UnitRecord | undefined`** — the ambient read. `undefined`
   outside a unit.
 - **`Clock` / `systemClock`** — `{ now, sleep(ms, signal?) }`, where `sleep`
@@ -720,7 +738,16 @@ orderActivities, workflows, imports })`, the sugar importing the starter;
 sync })`,
   `AmqpModule("OrderAmqpWorker")({ contract, handlers: orderHandlers, … })`),
   with its outbox relay a resourceful provider of its own rather than
-  something layered onto the runtime.
+  something layered onto the runtime. Both are also where **honouring the
+  kernel's deadline through the ambient record** is worked: neither middleware
+  injects anything into the call — `next()` unchanged — so
+  `currentUnit()?.signal` is the only route to it, and what each answers when
+  it is aborted is the transport's own business. `order-amqp-worker`'s
+  `orderChanged` returns a `RetryableError`, leaving the delivery un-acked so
+  the broker hands it to the next worker; `order-temporal-worker`'s
+  `ShippingService.arrange` fails as a **defect**, which the platform retries
+  on another worker — the contract's `ShippingUnavailable` is a permanent no
+  and would be the wrong error for "we ran out of time".
 - **`examples/order-api` consumes `@btravstack/http` rather than
   hand-rolling a transport, and its HTTP stack is the package's ONE way: oRPC
   over its own node adapter, `@unthrown/orpc` at the boundary.** The router is a di-provided

docs/examples/order-amqp-worker.md

@@ -49,6 +49,13 @@ export const orderHandlers = AmqpHandlers(orderContract)([Logger], {
   sync: (logger) => ({
     orderChanged: (message) => {
       const { id, payload } = message.payload;
+      if (currentUnit()?.signal.aborted === true) {
+        return ErrAsync(
+          new RetryableError(
+            `the drain deadline passed before order ${id} was notified`,
+          ),
+        );
+      }
       logger.info(
         payload === null
           ? `order ${id} is gone — notifying`
@@ -68,6 +75,14 @@ placement's `Err` never crosses the broker, only the committed fact does —
 which is why this deployment is absent from the `Err` table on the
 [overview](/examples/).
 
+The `currentUnit()?.signal` guard is the deployment's one kernel touchpoint,
+and it is how a handler honours the drain deadline at all: `messageUnits`
+calls `next()` unchanged, so there is no parameter to receive a signal
+through and the ambient record is the only route to it. Answering a
+`RetryableError` leaves the delivery **un-acked**, so the broker hands it to
+the next worker rather than this one finishing work nobody is waiting for.
+See [Read the ambient unit from an adapter](/how-to/read-the-ambient-unit).
+
 ## The relay: a resourceful provider with its own config
 
 The relay's one piece of configuration is a slice of this deployment's own,

docs/examples/order-temporal-worker.md

@@ -153,8 +153,35 @@ export const OrderTemporalWorker = TemporalModule("OrderTemporalWorker")({
 
 The same `ApplicationModule` + `PersistenceModule` pair as the API, plus
 `FulfillmentModule` — the two external services as in-memory stand-ins that
-always say yes, because what this deployment demonstrates is the
-orchestration; the specs swap in twins that say no. `TemporalModule` imports
+say yes to anything the drain still has time for, because what this deployment
+demonstrates is the orchestration; the specs swap in twins that say no.
+`ShippingService.arrange` is the exception, and the deployment's one kernel
+touchpoint:
+
+```ts
+arrange: (orderId) =>
+  currentUnit()?.signal.aborted === true
+    ? fromSafePromise(
+        Promise.reject(
+          new Error(
+            `the drain deadline passed before shipping for ${orderId} was arranged`,
+          ),
+        ),
+      )
+    : (logger.info(`arranged shipping for order ${orderId}`), OkAsync()),
+```
+
+An adapter is where reading the ambient record is legitimate, and here it is
+the only route to the unit's `AbortSignal` at all: `activityUnits` calls
+`next()` unchanged, so an activity has no parameter to receive one through.
+Failing as a **defect** is the point — the platform retries that attempt on
+another worker, where the contract's `ShippingUnavailable` is a permanent no
+and would be the wrong answer to "we ran out of time". Temporal's own
+`Context.current().cancellationSignal` is a different clock, firing on
+`shutdownGraceTime`; the two are honoured together. See
+[Read the ambient unit from an adapter](/how-to/read-the-ambient-unit).
+
+`TemporalModule` imports
 the starter (`TemporalRuntime`, `TemporalConfig` from `TEMPORAL_ADDRESS` /
 `TEMPORAL_NAMESPACE`, `TemporalConnection` as a resource of the graph),
 provides the activities and exports the runtime. `main.ts` is

docs/explanation/ambient-vs-context.md

@@ -1,6 +1,6 @@
 ---
 title: Ambient data, injected capabilities
-description: Why the kernel's AsyncLocalStorage record holds four fields of data and never a service, who is meant to read it, and the lint rule that does not exist yet.
+description: Why the kernel's AsyncLocalStorage record holds five fields of data and never a service, who is meant to read it, and the lint rule that does not exist yet.
 ---
 
 # Ambient data, injected capabilities
@@ -12,14 +12,15 @@ description: Why the kernel's AsyncLocalStorage record holds four fields of data
 > [`UnitMeta` and `UnitRecord`](/reference/core/runtime).
 
 The kernel opens one `AsyncLocalStorage` store per unit of work. It holds a
-small, fixed record — `UnitRecord`, four fields — and nothing else:
+small, fixed record — `UnitRecord`, five fields — and nothing else:
 
 ```ts
 type UnitRecord = {
   readonly unitId: string;
   readonly traceId: string;
   readonly tenantId: string | undefined;
   readonly deadline: number | undefined;
+  readonly signal: AbortSignal;
 };
 ```
 
@@ -53,7 +54,7 @@ So the line is: **ambient carries data, the di `Context` carries
 capabilities**. A repository through the store is the untestable coupling. A
 tenant id read by the Postgres adapter is not.
 
-## Why exactly these four
+## Why exactly these five
 
 Each field is a fact the kernel either mints or is handed at `run`, and each is
 one an adapter genuinely needs without being able to declare it.
@@ -71,10 +72,42 @@ one an adapter genuinely needs without being able to declare it.
   tagging a span — are exactly the readers the store is for.
 - `deadline` is a timestamp a runtime may pass so an adapter can budget a
   remote call against what is left.
+- `signal` is the **very** `AbortSignal` the kernel hands the unit's work
+  callback — one controller, two ways to reach it — fired at the drain
+  deadline, or at once on a path that skips the drain.
 
 There is no `Map`, no `set()`, no way for application code to add a field. A
 record that could grow would be a service locator with a smaller name.
 
+### Is a signal really data?
+
+It is the one field that looks like a capability, so it is worth saying why it
+is not. The test this page uses everywhere else is substitutability: a
+collaborator has an interface behind it, a test double to swap in, a
+`deps` array entry it should have been declared through. An `AbortSignal`
+has none of those. It is a fact about _this_ unit — "the process has stopped
+waiting for you" — exactly as `deadline` is the fact "this is when it will".
+Nothing about the code that reads it changes shape when it is absent; the
+`?.` in `currentUnit()?.signal` is the whole of the fallback.
+
+The reason it is on the record at all is that **the work callback is not
+always where the work is**. `@btravstack/http` opens the unit around its own
+listener, so it passes the signal as the handler's third parameter and never
+needs the record. `@btravstack/temporal` and `@btravstack/amqp` are
+middleware-shaped: the kernel's work callback is the library's `next()`, and
+an activity or a handler has no parameter to receive a signal through. The
+alternative was injecting a context — an extra first argument the Temporal or
+AMQP contract does not type — which is exactly the hidden-dependency shape
+this page argues against, so it was rejected. A deadline nobody can observe is
+not a deadline.
+
+A transport's own cancellation is a **different clock** and stays separate.
+Temporal's `Context.current().cancellationSignal` fires on a workflow-side
+cancellation, and on worker shutdown after `shutdownGraceTime`; AMQP has no
+cancellation story at all, since an un-acked delivery is redelivered, which is
+recovery rather than cancellation. The two are honoured together, not one
+standing in for the other.
+
 ## Who reads it
 
 Legitimate readers are **infrastructure adapters only** — the logger, the
@@ -101,7 +134,9 @@ The shipped runtimes are written to that rule. `@btravstack/http` opens a
 unit per request, `@btravstack/temporal` one per activity attempt and
 `@btravstack/amqp` one per delivery, and each injects nothing — a handler is a closure
 over the services its provider declared, and the ambient record is what an
-adapter underneath reads.
+adapter underneath reads. For the two middleware-shaped ones the record is
+also the only route to the unit's `AbortSignal`, since they call `next()`
+unchanged; `@btravstack/http` passes the same signal as an argument instead.
 
 ## The lint rule that does not exist
 

docs/explanation/draining-in-three-beats.md

@@ -81,7 +81,13 @@ connections, a Temporal worker finishing an activity — that window is wide. So
 `awaitIdle` is **sequenced behind** the runtime's `drain`, never alongside it.
 
 At the deadline, whatever is still open is aborted through each unit's
-`AbortSignal` and counted:
+`AbortSignal` — the one the kernel handed the work callback, and the one on
+the unit's ambient record as `currentUnit()?.signal`, which are the same
+object. Two routes matter because the callback is not always where the work
+is: a middleware-shaped runtime hands the kernel a callback that is the
+library's own `next()`, so a Temporal activity or an AMQP handler has no
+parameter to receive a signal through and reads it off the record instead.
+Whatever is aborted is then counted:
 
 ```ts
 type DrainReport = {

docs/how-to/consume-amqp-messages.md

@@ -175,6 +175,31 @@ id. A delivery tag is not used as the id on purpose: tags are per-channel and
 restart at `1` after the silent reconnects `amqp-connection-manager` performs.
 An adapter reads the trace id from `currentUnit()`.
 
+## Honouring the drain deadline
+
+The middleware calls `next()` unchanged, so a handler has no parameter to
+receive the unit's `AbortSignal` through: `currentUnit()?.signal` is the only
+route to it, and it is aborted at the kernel's `drainTimeoutMs`.
+
+```ts
+orderChanged: (message) => {
+  const { id, payload } = message.payload;
+  if (currentUnit()?.signal.aborted === true) {
+    return ErrAsync(
+      new RetryableError(
+        `the drain deadline passed before order ${id} was notified`,
+      ),
+    );
+  }
+  // …
+};
+```
+
+A `RetryableError` leaves the delivery **un-acked**, so the broker hands it to
+the next worker — the transport's own answer to "this process stopped waiting".
+There is no cancellation to defer to here: AMQP has none, and a redelivery is
+recovery rather than cancellation.
+
 ## The drain: one deadline
 
 `Serving.drain(signal)` calls `worker.close({ drainTimeoutMs: null })` —