diff --git a/packages/context/src/context/layer-lifecycle.ts b/packages/context/src/context/layer-lifecycle.ts
index 48500162..0580bbd2 100644
--- a/packages/context/src/context/layer-lifecycle.ts
+++ b/packages/context/src/context/layer-lifecycle.ts
@@ -480,8 +480,14 @@ export async function initLayers({ layers, ctx, storage, store }: InitLayersPara
  * and `store()` hooks — are mirrored durably; `store()` is not special.
  * 'execution' scope is excluded: its scope key rotates each run, so there is
  * nothing durable to mirror.
+ *
+ * Exported (rather than private to `initLayers`) because a host that carries
+ * layer state forward across executions — warm hydration, see
+ * `AgentHarness.ensureLayersInit` — has to re-point write-through at the new
+ * executionId WITHOUT re-running `init`, which is the whole point of carrying
+ * state forward.
  */
-function registerDurableTargets({
+export function registerDurableTargets({
   layers,
   ctx,
   storage,
diff --git a/packages/core/src/harness/agent-harness.ts b/packages/core/src/harness/agent-harness.ts
index 2f90ad29..00acf271 100644
--- a/packages/core/src/harness/agent-harness.ts
+++ b/packages/core/src/harness/agent-harness.ts
@@ -26,7 +26,9 @@ import {
   recallLayers,
   recallLayersAtomic,
   recallLayersEventual,
+  registerDurableTargets,
   resolveLayerTools,
+  resolveScopeKey,
   runAppendPipeline,
   storeLayers,
 } from './deps/context';
@@ -57,6 +59,7 @@ import {
   createStepLedgerStore,
   filterReasoningStream,
   filterTextStream,
+  ItemLogImpl,
   resolveStepLedgerRetention,
   restoreFromCheckpoint,
   SessionRunner,
@@ -206,7 +209,14 @@ interface AgentHarnessOpts<TParams extends Record<string, unknown> = Record<stri
 
 interface Session {
   readonly runner: SessionRunner;
-  accumulatedItems: Item[];
+  /**
+   * The single session-owned conversation log. Every turn's context shares
+   * this instance by reference — there is no per-turn copy-forward or
+   * copy-back, and `previewRequestItems` reads the same object the turn
+   * writes. Failed/aborted turns are rolled back to a watermark by the
+   * runner so they leave no trace, preserving the old copy semantics.
+   */
+  readonly log: ItemLogImpl;
 }
 
 //#endregion
@@ -455,6 +465,26 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
    * `ctx.id` (the layer-state store's executionId).
    */
   private readonly initializedExecutions = new Set<string>();
+  /**
+   * `<layerId>@<scopeKey>` → the execution that last hydrated that layer's state
+   * (warm layer-state carry-forward).
+   *
+   * Keyed by BUCKET, not by thread. A layer's state lives under
+   * `resolveScopeKey(layer.scope, ctx)`, and for 'resource' and 'global' scope
+   * that key is not a function of the thread: two turns on one thread with
+   * different `resourceId`s address different resource buckets, and every thread
+   * in the process shares the one global bucket. Keying on thread identity gets
+   * both wrong in opposite directions — it carries state ACROSS resource buckets
+   * (one tenant's state into another's, then persisted there, because
+   * `registerDurableTargets` re-points write-through at the new scope key), and
+   * it FAILS to carry state across threads sharing the global bucket (each
+   * thread resumes its own stale copy, so the write-through mirror makes the last
+   * writer win and drops the others' updates).
+   *
+   * Keying by bucket makes both a function of the same fact: a layer continues
+   * from whatever execution last touched the bucket it is about to read.
+   */
+  private readonly hydratedLayers = new Map<string, string>();
   readonly traceExporter: TraceExporter;
   /**
    * Long-lived shared cwd state. The same reference is seeded into every
@@ -466,6 +496,13 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
    *  in `runtime/durable/` (restore) can parse persisted items. Do not
    *  access from outside core. */
   readonly itemSchemas: ItemSchemaRegistry;
+  /**
+   * Memoized harness-base ∪ harness-layer item registry, used for the shared
+   * session log. `_contextLayers` is readonly and set once in the constructor,
+   * so the extension is a pure function of construction options — see
+   * `sessionItemSchemas`.
+   */
+  private _sessionItemSchemasCache?: ItemSchemaRegistry;
 
   constructor(opts: AgentHarnessOpts<TParams>) {
     const validatedParams = opts.paramsSchema ? opts.paramsSchema.parse(opts.params) : opts.params;
@@ -635,9 +672,16 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
 
   seedSessionHistory(threadId: string, items: ReadonlyArray<Item>): void {
     const session = this.getOrCreateSession(threadId);
-    session.accumulatedItems = [
-      ...items,
-    ];
+    // Validate the entire replacement first so one bad item cannot destroy
+    // existing history or leave a partially seeded shared log.
+    const validated = new ItemLogImpl(this.sessionItemSchemas());
+    for (const item of items) {
+      validated.append(item);
+    }
+    session.log.truncateTo(0);
+    for (const item of validated.items) {
+      session.log.append(item);
+    }
   }
 
   private getOrCreateSession(threadId: string): Session {
@@ -646,8 +690,17 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
       return existing;
     }
 
+    // The LAYER-EXTENDED registry, not the harness base. The session log is
+    // shared with every context this thread builds, and `createContext` extends
+    // the base with the layers' `itemSchemas` — so a log bound to the base would
+    // reject exactly the custom item types those layers declare, on both the
+    // `seedSessionHistory` path and mid-turn (`onItemAppend`, tool results).
+    const sessionLog = new ItemLogImpl(this.sessionItemSchemas());
+    // Watermark captured at turn start (before the turn's input lands); a
+    // failed turn truncates back to it so partial items leave no trace.
+    let turnWatermark = 0;
     const session: Session = {
-      accumulatedItems: [],
+      log: sessionLog,
       runner: new SessionRunner({
         threadId,
         agentName: this.config.name,
@@ -658,12 +711,14 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
         // time and doesn't apply here.
         createContext: (items, _turnId, messages) => {
           const perTurnOptions: ExecuteOptions = messages[0]?.options ?? {};
-          const allItems: Item[] = [
-            ...session.accumulatedItems,
-            ...items,
-          ];
+          // Append this turn's input to the SESSION log and hand the same log
+          // to the context — single owner, zero per-turn copies.
+          turnWatermark = sessionLog.length;
+          for (const item of items) {
+            sessionLog.append(item);
+          }
           const ctx = this.createContext({
-            items: allItems,
+            itemLog: sessionLog,
             threadId,
             resourceId: perTurnOptions.resourceId,
             state: perTurnOptions.state,
@@ -683,6 +738,9 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
           }
           return ctx;
         },
+        rollbackTurn: () => {
+          sessionLog.truncateTo(turnWatermark);
+        },
         runTurn: async (ctx, _turn, signal) => {
           if (!this.agentGraph) {
             throw new NoeticConfigError({
@@ -706,10 +764,6 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
             );
           }
           const result = await this.initAndRun(this.agentGraph, '', ctx);
-          // Snapshot final items into session history for the next turn.
-          session.accumulatedItems = [
-            ...ctx.itemLog.items,
-          ];
           return result;
         },
       }),
@@ -815,8 +869,19 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
    * treated as disabled. Guarded so nested/repeated `run()` calls and the
    * session turn path never re-init — re-init would re-hydrate from storage and
    * clobber accumulated in-memory state.
+   *
+   * Turns addressing the same layer scope bucket take the WARM path: layer
+   * state carries forward in-memory rather than being re-read from storage
+   * (sequential reads, each with a 10s timeout, every turn). `transient: true`
+   * opts a throwaway context out of publishing itself as a warm source — see
+   * `previewRequestItems`, which tears its execution's state down again.
    */
-  private async ensureLayersInit(ctx: Context): Promise<void> {
+  private async ensureLayersInit(
+    ctx: Context,
+    opts?: {
+      transient?: boolean;
+    },
+  ): Promise<void> {
     const layers = ctx.layers;
     const storage = this.config.storage;
     if (!layers || layers.length === 0 || !storage) {
@@ -826,7 +891,162 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
       return;
     }
     this.initializedExecutions.add(ctx.id);
-    await this.initLayers(layers, ctx, storage);
+    const warmKeys = this.resolveWarmKeys(layers, ctx);
+    await this.tryWarmInit({
+      warmKeys,
+      layers,
+      ctx,
+      storage,
+    });
+    if (!opts?.transient) {
+      for (const [layerId, warmKey] of warmKeys) {
+        // Only publish for layers this execution actually holds state for; a
+        // layer whose init failed with `onInitError: 'disable'` has nothing to
+        // hand the next turn.
+        if (this.layerStateStore.has?.(ctx.id, layerId)) {
+          this.hydratedLayers.set(warmKey, ctx.id);
+        }
+      }
+    }
+  }
+
+  /**
+   * The warm-cache key for each layer: its id plus the storage bucket it
+   * resolves to for `ctx`. Execution-scoped layers are omitted — they never
+   * carry forward, and their scope key is `ctx.id`, so an entry would be a
+   * per-run leak.
+   */
+  private resolveWarmKeys(
+    layers: ReadonlyArray<ContextLayer>,
+    ctx: Context,
+  ): ReadonlyMap<string, string> {
+    const execCtx = this.toExecCtx(ctx);
+    const keys = new Map<string, string>();
+    for (const layer of layers) {
+      if (layer.scope === 'execution') {
+        continue;
+      }
+      keys.set(layer.id, `${layer.id}@${resolveScopeKey(layer.scope, execCtx)}`);
+    }
+    return keys;
+  }
+
+  /**
+   * Warm path: some previous execution already hydrated these layers from
+   * storage. Copy the live in-memory state forward to the new executionId
+   * instead of re-running every init. The state store is the source of truth
+   * between turns — its durable write-through keeps storage in sync.
+   *
+   * Resolved PER LAYER against `(layer, scopeKey)`, which is the identity of the
+   * storage bucket the layer's state actually lives in. A layer therefore carries
+   * forward from the last execution that touched ITS bucket, whatever thread that
+   * was: a resource-scoped layer on a turn with a new `resourceId` finds no entry
+   * for the new bucket and cold-inits from it, while a global-scoped layer shares
+   * one entry across every thread, so an increment on thread B is what thread A's
+   * next turn continues from. Execution-scoped layers are per-run by contract and
+   * never carry forward.
+   *
+   * Every layer this did not carry forward is cold-inited here, so the caller
+   * needs no fallback — the return value is informational.
+   */
+  private async tryWarmInit({
+    warmKeys,
+    layers,
+    ctx,
+    storage,
+  }: {
+    warmKeys: ReadonlyMap<string, string>;
+    layers: ContextLayer[];
+    ctx: Context;
+    storage: StorageAdapter;
+  }): Promise<boolean> {
+    const carried = this.carryLayerStateForward({
+      warmKeys,
+      layers,
+      ctx,
+    });
+    // Cold-init covers execution-scoped layers, layers whose bucket has no warm
+    // entry, and layers that were never hydrated in the first place.
+    const cold = layers.filter(
+      (l) => l.scope === 'execution' || !this.layerStateStore.has?.(ctx.id, l.id),
+    );
+    if (cold.length > 0) {
+      await this.initLayers(cold, ctx, storage);
+    }
+    if (carried === 0) {
+      // `initLayers` already registered durable targets for every layer it ran,
+      // which — with nothing carried — is all of them.
+      return false;
+    }
+    // Re-point write-through at the new execution id for the carried layers too,
+    // so their state keeps mirroring durably without a re-`init`.
+    registerDurableTargets({
+      layers: layers.filter((l) => l.scope !== 'execution'),
+      ctx: this.toExecCtx(ctx),
+      storage,
+      store: this.layerStateStore,
+    });
+    return true;
+  }
+
+  /**
+   * Copy each layer's live state forward from whichever execution last hydrated
+   * that layer's bucket. Returns how many layers were carried.
+   */
+  private carryLayerStateForward({
+    warmKeys,
+    layers,
+    ctx,
+  }: {
+    warmKeys: ReadonlyMap<string, string>;
+    layers: ReadonlyArray<ContextLayer>;
+    ctx: Context;
+  }): number {
+    let copied = 0;
+    for (const layer of layers) {
+      const warmKey = warmKeys.get(layer.id);
+      if (warmKey === undefined) {
+        // Execution scope — `resolveWarmKeys` omits it.
+        continue;
+      }
+      const warm = this.hydratedLayers.get(warmKey);
+      // `warm === ctx.id` is a re-entrant init of the same execution: there is
+      // nothing to copy, and copying onto itself would be a no-op anyway.
+      if (warm === undefined || warm === ctx.id) {
+        continue;
+      }
+      if (!this.layerStateStore.has?.(warm, layer.id)) {
+        // The warm execution's state was torn down (dispose/cleanup). Drop the
+        // stale pointer so later turns stop probing it.
+        this.hydratedLayers.delete(warmKey);
+        continue;
+      }
+      this.layerStateStore.set(ctx.id, layer.id, this.layerStateStore.get(warm, layer.id));
+      copied++;
+    }
+    return copied;
+  }
+
+  /**
+   * Item registry for the SHARED SESSION LOG: the harness base extended with
+   * the harness-level context layers' `itemSchemas`.
+   *
+   * Deliberately keyed to the HARNESS layer set, not a turn's. `createContext`
+   * lets a caller pass per-turn `contextLayers`, and those layers' item types
+   * are validated by that context's own (wider) registry on the paths that build
+   * one — but the log outlives any single turn and is shared by all of them, so
+   * binding it to one turn's layer set would make what the log accepts depend on
+   * whichever turn happened to create the session. Harness-level layers are the
+   * stable set every turn on the thread has, so they are the log's contract; a
+   * per-turn layer that declares a brand-new item type and appends it to the
+   * shared log must also be declared at harness level.
+   */
+  private sessionItemSchemas(): ItemSchemaRegistry {
+    this._sessionItemSchemasCache ??= buildItemSchemaRegistry({
+      base: this.itemSchemas,
+      layers: this._contextLayers,
+    });
+    return this._sessionItemSchemasCache;
   }
 
   detachedSpawn<I, O>(
@@ -845,6 +1065,8 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
   createContext(opts?: {
     parent?: Context;
     items?: Item[];
+    /** @internal Share the session-owned log instead of seeding from `items`. */
+    itemLog?: ItemLogImpl;
     state?: unknown;
     threadId?: string;
     resourceId?: string;
@@ -1045,7 +1267,7 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
     const existingSession = this.sessions.get(threadId);
     const historyItems: Item[] = existingSession
       ? [
-          ...existingSession.accumulatedItems,
+          ...existingSession.log.items,
         ]
       : [];
     const ctx = this.createContext({
@@ -1058,7 +1280,14 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
       return historyItems;
     }
     try {
-      await this.ensureLayersInit(ctx);
+      /* `transient`: this context's state is torn down in the `finally` below, so
+       * it must NOT become a warm-hydration source — a real turn that followed
+       * it would find a pointer to wiped state, carry nothing forward, and
+       * silently cold-init (correct, but the warm win evaporates after any
+       * preview, which a TUI may issue on every keystroke). */
+      await this.ensureLayersInit(ctx, {
+        transient: true,
+      });
       const recallResults = await this.recallLayers(layers, '', ctx);
       // Band the output the way a real turn would, but read-only: a preview
       // must not pin, count churn, or age the epoch, or looking at a
@@ -1116,6 +1345,14 @@ export class AgentHarness<TParams extends Record<string, unknown> = Record<strin
     // Drop the init guard so a deliberate dispose→run cycle re-hydrates from
     // storage (disposeLayers also clears the layer-state store for this id).
     this.initializedExecutions.delete(ctx.id);
+    // ...and any warm pointer aimed at this execution, whose state is about to
+    // be wiped. A stale pointer only degrades to a cold init, but dropping it
+    // here keeps the two guards from disagreeing about what is hydrated.
+    for (const [warmKey, executionId] of this.hydratedLayers) {
+      if (executionId === ctx.id) {
+        this.hydratedLayers.delete(warmKey);
+      }
+    }
     await disposeLayers({
       layers,
       ctx: this.toExecCtx(ctx),
diff --git a/packages/core/src/harness/deps/context.ts b/packages/core/src/harness/deps/context.ts
index aad3b225..99d93df9 100644
--- a/packages/core/src/harness/deps/context.ts
+++ b/packages/core/src/harness/deps/context.ts
@@ -17,7 +17,9 @@ export {
   recallLayers,
   recallLayersAtomic,
   recallLayersEventual,
+  registerDurableTargets,
   resolveLayerTools,
+  resolveScopeKey,
   runAppendPipeline,
   storeLayers,
 } from '@noetic-tools/context';
diff --git a/packages/core/src/harness/deps/runtime.ts b/packages/core/src/harness/deps/runtime.ts
index 526857a6..3eeab957 100644
--- a/packages/core/src/harness/deps/runtime.ts
+++ b/packages/core/src/harness/deps/runtime.ts
@@ -20,6 +20,7 @@ export {
 } from '../../runtime/durable/step-ledger';
 export type { EventBroadcaster } from '../../runtime/event-broadcaster';
 export { createInMemoryStorage } from '../../runtime/in-memory-storage';
+export { ItemLogImpl } from '../../runtime/item-log-impl';
 export type { QueuedMessage } from '../../runtime/message-queue';
 export { SessionRunner } from '../../runtime/session-runner';
 export {
diff --git a/packages/core/src/runtime/context-impl.ts b/packages/core/src/runtime/context-impl.ts
index 24c59628..0ecea64a 100644
--- a/packages/core/src/runtime/context-impl.ts
+++ b/packages/core/src/runtime/context-impl.ts
@@ -155,6 +155,13 @@ export class ContextImpl implements Context<ContextData> {
     harness: AgentHarnessContract;
     parent?: Context;
     items?: Item[];
+    /**
+     * Share an existing log instead of building a fresh one from `items`.
+     * The session runner passes its session-owned log here so every turn in a
+     * thread appends to ONE log — no copy-forward/copy-back per turn.
+     * Mutually exclusive with `items`.
+     */
+    itemLog?: ItemLogImpl;
     state?: unknown;
     threadId?: string;
     resourceId?: string;
@@ -200,13 +207,17 @@ export class ContextImpl implements Context<ContextData> {
     };
     this._broadcaster = opts._broadcaster;
 
-    const log = new ItemLogImpl(this.itemSchemas);
-    if (opts.items) {
-      for (const item of opts.items) {
-        log.append(item);
+    if (opts.itemLog) {
+      this.itemLog = opts.itemLog;
+    } else {
+      const log = new ItemLogImpl(this.itemSchemas);
+      if (opts.items) {
+        for (const item of opts.items) {
+          log.append(item);
+        }
       }
+      this.itemLog = log;
     }
-    this.itemLog = log;
 
     // Join the parent's abort cascade last, so the child is fully constructed
     // before an already-aborted parent aborts it.
diff --git a/packages/core/src/runtime/item-log-impl.ts b/packages/core/src/runtime/item-log-impl.ts
index baf42415..eecf606e 100644
--- a/packages/core/src/runtime/item-log-impl.ts
+++ b/packages/core/src/runtime/item-log-impl.ts
@@ -24,4 +24,23 @@ export class ItemLogImpl implements ItemLog {
     this._items.push(this.itemSchemas.parse(item));
     this._frozenCache = null;
   }
+
+  /** @internal Current length — used as a rollback watermark by the session runner. */
+  get length(): number {
+    return this._items.length;
+  }
+
+  /**
+   * @internal Roll the log back to a previously-captured watermark. Used ONLY
+   * by the session runner to discard a failed/aborted turn's partial items so
+   * a shared session log preserves the same "failed turns leave no trace"
+   * semantics the copy-based history had.
+   */
+  truncateTo(watermark: number): void {
+    if (watermark < 0 || watermark >= this._items.length) {
+      return;
+    }
+    this._items.length = watermark;
+    this._frozenCache = null;
+  }
 }
diff --git a/packages/core/src/runtime/session-runner.ts b/packages/core/src/runtime/session-runner.ts
index a2a46f2a..4d4eb4b6 100644
--- a/packages/core/src/runtime/session-runner.ts
+++ b/packages/core/src/runtime/session-runner.ts
@@ -42,6 +42,13 @@ export interface SessionRunnerOpts {
   readonly agentName: string;
   readonly runTurn: RunTurnFn;
   readonly createContext: CreateContextFn;
+  /**
+   * Roll back session-owned state after a failed/aborted turn. With a shared
+   * session log (single-owner history), a failed turn's partial items must be
+   * discarded explicitly to preserve the "failed turns leave no trace"
+   * contract the old copy-back gave for free.
+   */
+  readonly rollbackTurn?: () => void;
 }
 
 //#endregion
@@ -111,6 +118,7 @@ export class SessionRunner {
   private readonly agentName: string;
   private readonly runTurn: RunTurnFn;
   private readonly createContext: CreateContextFn;
+  private readonly rollbackTurn?: () => void;
 
   private status: HarnessStatus = {
     kind: 'idle',
@@ -139,6 +147,7 @@ export class SessionRunner {
     this.agentName = opts.agentName;
     this.runTurn = opts.runTurn;
     this.createContext = opts.createContext;
+    this.rollbackTurn = opts.rollbackTurn;
 
     this.queue.subscribe(() => {
       this.kick();
@@ -245,37 +254,39 @@ export class SessionRunner {
     // the harness's createContext callback receives Item[] directly and
     // seeds the context without re-converting.
     const items = mergeInputsToItems(messages);
-    const ctx = this.createContext(items, turnId, messages);
-    this.currentCtx = ctx;
-
-    emitFrameworkEvent({
-      broadcaster: this.broadcaster,
-      agentName: this.agentName,
-      eventType: 'turn_started',
-      data: {
-        turnId,
-        messageIds: messages.map((m) => m.id),
-      },
-    });
-    // Input items never appear in the SDK stream — item_appended is what
-    // carries them into getItemStream (and any other log-faithful consumer).
-    for (const item of items) {
+    let ctx: Context | undefined;
+
+    try {
+      ctx = this.createContext(items, turnId, messages);
+      this.currentCtx = ctx;
+
       emitFrameworkEvent({
         broadcaster: this.broadcaster,
         agentName: this.agentName,
-        eventType: 'item_appended',
+        eventType: 'turn_started',
         data: {
-          item,
+          turnId,
+          messageIds: messages.map((m) => m.id),
         },
       });
-    }
+      // Input items never appear in the SDK stream — item_appended is what
+      // carries them into getItemStream (and any other log-faithful consumer).
+      for (const item of items) {
+        emitFrameworkEvent({
+          broadcaster: this.broadcaster,
+          agentName: this.agentName,
+          eventType: 'item_appended',
+          data: {
+            item,
+          },
+        });
+      }
 
-    const turn: TurnContext = {
-      turnId,
-      session: this,
-    };
+      const turn: TurnContext = {
+        turnId,
+        session: this,
+      };
 
-    try {
       const text = await this.runTurn(ctx, turn, controller.signal);
       const response = buildResponse(text, ctx);
       this.lastResponse = response;
@@ -299,6 +310,7 @@ export class SessionRunner {
     } catch (err: unknown) {
       const error = err instanceof Error ? err : new Error(String(err));
       this.lastError = error;
+      this.rollbackTurn?.();
       emitFrameworkEvent({
         broadcaster: this.broadcaster,
         agentName: this.agentName,
@@ -312,14 +324,16 @@ export class SessionRunner {
     } finally {
       // Accumulate the turn's token accounting whatever the outcome — an
       // aborted turn still consumed whatever the model billed before the cut.
-      this.totalInputTokens += ctx.tokens.input;
-      this.totalOutputTokens += ctx.tokens.output;
-      // Only ever leave `undefined` behind when NO turn reported a figure —
-      // a turn that reported 0 must surface as 0, not "unreported".
-      if (ctx.tokens.cached !== undefined) {
-        this.totalCachedTokens = (this.totalCachedTokens ?? 0) + ctx.tokens.cached;
+      if (ctx) {
+        this.totalInputTokens += ctx.tokens.input;
+        this.totalOutputTokens += ctx.tokens.output;
+        // Only ever leave `undefined` behind when NO turn reported a figure —
+        // a turn that reported 0 must surface as 0, not "unreported".
+        if (ctx.tokens.cached !== undefined) {
+          this.totalCachedTokens = (this.totalCachedTokens ?? 0) + ctx.tokens.cached;
+        }
+        this.totalCost += ctx.cost;
       }
-      this.totalCost += ctx.cost;
       this.currentCtx = undefined;
       this.currentController = undefined;
       this.status = {
diff --git a/packages/core/test/context/warm-layer-hydration.test.ts b/packages/core/test/context/warm-layer-hydration.test.ts
new file mode 100644
index 00000000..63dbd957
--- /dev/null
+++ b/packages/core/test/context/warm-layer-hydration.test.ts
@@ -0,0 +1,194 @@
+/**
+ * Warm layer hydration: successive turns addressing the same layer scope bucket
+ * carry non-execution-scoped layer state forward IN MEMORY instead of re-reading
+ * it from storage.
+ *
+ * Cold-initing every turn cost one sequential storage read per layer — each with
+ * its own 10s timeout — on every single turn, for state the state store already
+ * held. Execution-scoped layers are exempt: their scope key rotates per run, so
+ * they must still init per turn by contract.
+ */
+
+import { describe, expect, it } from 'bun:test';
+import type { ContextData, ContextLayer } from '@noetic-tools/context';
+import type { LLMResponse, MessageItem, Step } from '@noetic-tools/types';
+import { AgentHarness } from '../../src/harness/agent-harness';
+import { assistantMessage, makeStorage } from '../_helpers';
+
+interface CountState {
+  count: number;
+}
+
+interface Probe {
+  readonly layer: ContextLayer<CountState>;
+  readonly inits: () => number;
+}
+
+/** A layer that counts how many times `init` ran, and bumps state per turn. */
+function countingLayer(scope: 'thread' | 'execution'): Probe {
+  let inits = 0;
+  const layer: ContextLayer<CountState> = {
+    id: `counter-${scope}`,
+    slot: 100,
+    scope,
+    hooks: {
+      async init({ storage }) {
+        inits += 1;
+        const saved = await storage.get<CountState>('state');
+        return {
+          state: saved ?? {
+            count: 0,
+          },
+        };
+      },
+      async recall({ state }) {
+        return `count=${state.count}`;
+      },
+      async store({ state }) {
+        return {
+          state: {
+            count: (state?.count ?? 0) + 1,
+          },
+        };
+      },
+    },
+  };
+  return {
+    layer,
+    inits: () => inits,
+  };
+}
+
+const chatStep: Step<ContextData, string, string> = {
+  kind: 'callModel',
+  id: 'chat',
+  model: 'test/scripted',
+  tools: [],
+};
+
+function harnessWith(layers: ContextLayer[]): AgentHarness {
+  let call = 0;
+  return new AgentHarness({
+    name: 'warm-test',
+    params: {},
+    agentGraph: chatStep,
+    environment: {
+      storage: {
+        adapter: makeStorage(),
+      },
+    },
+    contextLayers: layers,
+    _testCallModel: async (): Promise<LLMResponse> => {
+      const message: MessageItem = assistantMessage(`answer ${call}`, `resp-${call}`);
+      call += 1;
+      return {
+        items: [
+          message,
+        ],
+        usage: {
+          inputTokens: 0,
+          outputTokens: 1,
+        },
+      };
+    },
+  });
+}
+
+async function runTurns(harness: AgentHarness, threadId: string, count: number): Promise<void> {
+  for (let i = 0; i < count; i++) {
+    await harness.execute(`turn ${i}`, {
+      threadId,
+    });
+    await harness.getAgentResponse({
+      threadId,
+    });
+  }
+}
+
+describe('warm layer hydration across turns on one thread', () => {
+  it('a thread-scoped layer inits once per thread, not once per turn', async () => {
+    const probe = countingLayer('thread');
+    const harness = harnessWith([
+      probe.layer,
+    ]);
+    await runTurns(harness, 'warm', 3);
+    expect(probe.inits()).toBe(1);
+  });
+
+  it('carried-forward state keeps accumulating across turns', async () => {
+    const probe = countingLayer('thread');
+    const harness = harnessWith([
+      probe.layer,
+    ]);
+    await runTurns(harness, 'warm', 3);
+    // `store` bumps the counter once per turn, and the warm path hands the live
+    // state to the next turn — so a cold re-init would reset visible progress.
+    const items = await harness.previewRequestItems({
+      threadId: 'warm',
+    });
+    const texts = items.flatMap((item) =>
+      item.type === 'message'
+        ? item.content.flatMap((part) =>
+            part.type === 'input_text' || part.type === 'output_text'
+              ? [
+                  part.text,
+                ]
+              : [],
+          )
+        : [],
+    );
+    expect(texts.some((t) => t.includes('count=3'))).toBe(true);
+  });
+
+  it('an execution-scoped layer still inits every turn', async () => {
+    const probe = countingLayer('execution');
+    const harness = harnessWith([
+      probe.layer,
+    ]);
+    await runTurns(harness, 'warm', 3);
+    expect(probe.inits()).toBe(3);
+  });
+
+  it('separate threads hydrate independently', async () => {
+    const probe = countingLayer('thread');
+    const harness = harnessWith([
+      probe.layer,
+    ]);
+    await runTurns(harness, 'thread-a', 2);
+    await runTurns(harness, 'thread-b', 2);
+    // One cold init per thread — the scope key is thread-scoped, so thread B
+    // cannot ride thread A's carry-forward.
+    expect(probe.inits()).toBe(2);
+  });
+
+  it('a preview between turns does not poison the thread warm pointer', async () => {
+    /* `previewRequestItems` inits layers on a THROWAWAY context and tears that
+     * execution's state down in its `finally`. If the preview published itself as
+     * the bucket's warm source, the next real turn would find a pointer to wiped
+     * state, carry nothing forward, and silently cold-init — correct, but the
+     * warm win would evaporate after any preview (a TUI may issue one per
+     * keystroke). */
+    const probe = countingLayer('thread');
+    const harness = harnessWith([
+      probe.layer,
+    ]);
+    await runTurns(harness, 'warm', 1);
+    expect(probe.inits()).toBe(1);
+
+    // Previews themselves ride the warm path off turn 1, so they add no inits...
+    await harness.previewRequestItems({
+      threadId: 'warm',
+    });
+    await harness.previewRequestItems({
+      threadId: 'warm',
+    });
+    expect(probe.inits()).toBe(1);
+
+    // ...and, crucially, they did not become the bucket's warm SOURCE. Turn 2
+    // still resolves to turn 1's live state. Were the preview the source, its
+    // `finally` (flush + cleanup) would have wiped that execution's state, the
+    // warm copy would find nothing, and this turn would cold-init to 2.
+    await runTurns(harness, 'warm', 1);
+    expect(probe.inits()).toBe(1);
+  });
+});
diff --git a/packages/core/test/runtime/context-impl.test.ts b/packages/core/test/runtime/context-impl.test.ts
index befe8fae..3f13baae 100644
--- a/packages/core/test/runtime/context-impl.test.ts
+++ b/packages/core/test/runtime/context-impl.test.ts
@@ -4,6 +4,7 @@ import { isNoeticError } from '@noetic-tools/types';
 import { z } from 'zod';
 import { ChannelStore } from '../../src/runtime/channel-store';
 import { ContextImpl, collectContextTree } from '../../src/runtime/context-impl';
+import { ItemLogImpl } from '../../src/runtime/item-log-impl';
 import { makeMockContext, makeMockHarness } from '../_helpers';
 
 function makeTestItem(): InputMessageItem {
@@ -474,3 +475,49 @@ describe('collectContextTree', () => {
     ]);
   });
 });
+
+describe('ContextImpl shared itemLog option', () => {
+  /**
+   * The session runner hands every turn's context the ONE session-owned log.
+   * Appending through the context must land on the shared instance, and reads
+   * through either handle must observe the other's writes — that identity is
+   * what replaces the old copy-forward/copy-back history.
+   */
+  test('a context constructed with `itemLog` shares the instance by reference', () => {
+    const shared = new ItemLogImpl();
+    shared.append(makeTestItem());
+    const ctx = new ContextImpl({
+      harness: makeMockHarness(),
+      itemLog: shared,
+    });
+
+    expect(ctx.itemLog).toBe(shared);
+    expect(ctx.itemLog.items).toHaveLength(1);
+
+    // Writes through the context land on the shared log...
+    ctx.itemLog.append({
+      ...makeTestItem(),
+      id: 'item-2',
+    });
+    expect(shared.items).toHaveLength(2);
+
+    // ...and a rollback through the shared handle is visible to the context.
+    shared.truncateTo(1);
+    expect(ctx.itemLog.items).toHaveLength(1);
+  });
+
+  test('without `itemLog`, a context still builds its own log from `items`', () => {
+    const ctx = new ContextImpl({
+      harness: makeMockHarness(),
+      items: [
+        makeTestItem(),
+      ],
+    });
+    expect(ctx.itemLog.items).toHaveLength(1);
+    // Mutating the context's log must not reach into any other instance.
+    const other = new ContextImpl({
+      harness: makeMockHarness(),
+    });
+    expect(other.itemLog.items).toHaveLength(0);
+  });
+});
diff --git a/packages/core/test/runtime/item-log-impl.test.ts b/packages/core/test/runtime/item-log-impl.test.ts
index 37b4bcae..681075d4 100644
--- a/packages/core/test/runtime/item-log-impl.test.ts
+++ b/packages/core/test/runtime/item-log-impl.test.ts
@@ -143,3 +143,83 @@ describe('ItemLogImpl', () => {
     expect(log.items[4].type).toBe('openrouter:datetime');
   });
 });
+
+describe('ItemLogImpl.length + truncateTo (session rollback watermark)', () => {
+  const seed = (count: number): ItemLogImpl => {
+    const log = new ItemLogImpl();
+    for (let i = 0; i < count; i++) {
+      log.append(makeInputMessage(`m${i}`));
+    }
+    return log;
+  };
+
+  it('length tracks appends', () => {
+    const log = new ItemLogImpl();
+    expect(log.length).toBe(0);
+    log.append(makeInputMessage('m1'));
+    expect(log.length).toBe(1);
+    log.append(makeInputMessage('m2'));
+    expect(log.length).toBe(2);
+  });
+
+  it('truncateTo drops every item at or after the watermark', () => {
+    const log = seed(3);
+    log.truncateTo(1);
+    expect(log.length).toBe(1);
+    expect(log.items.map((i) => ('id' in i ? i.id : ''))).toEqual([
+      'm0',
+    ]);
+  });
+
+  it('truncateTo(0) empties the log', () => {
+    const log = seed(2);
+    log.truncateTo(0);
+    expect(log.length).toBe(0);
+    expect(log.items).toEqual([]);
+  });
+
+  it('invalidates the frozen snapshot so a later read sees the truncation', () => {
+    const log = seed(2);
+    // Materialise the cache first — a stale cache would keep returning 2 items.
+    expect(log.items).toHaveLength(2);
+    log.truncateTo(1);
+    expect(log.items).toHaveLength(1);
+  });
+
+  // Boundary sweep at N-1 / N / N+1 around the current length (N = 3): only
+  // watermarks strictly below the length truncate; >= length and negative are
+  // no-ops so a stale/out-of-range watermark can never grow or corrupt the log.
+  it('truncateTo(length - 1) truncates', () => {
+    const log = seed(3);
+    log.truncateTo(2);
+    expect(log.length).toBe(2);
+  });
+
+  it('truncateTo(length) is a no-op', () => {
+    const log = seed(3);
+    log.truncateTo(3);
+    expect(log.length).toBe(3);
+  });
+
+  it('truncateTo(length + 1) is a no-op', () => {
+    const log = seed(3);
+    log.truncateTo(4);
+    expect(log.length).toBe(3);
+  });
+
+  it('truncateTo(negative) is a no-op', () => {
+    const log = seed(3);
+    log.truncateTo(-1);
+    expect(log.length).toBe(3);
+  });
+
+  it('appending after a truncation continues from the watermark', () => {
+    const log = seed(3);
+    log.truncateTo(1);
+    log.append(makeInputMessage('after'));
+    expect(log.items.map((i) => ('id' in i ? i.id : ''))).toEqual([
+      'm0',
+      'after',
+    ]);
+  });
+});
diff --git a/packages/core/test/runtime/session-layer-identity.test.ts b/packages/core/test/runtime/session-layer-identity.test.ts
new file mode 100644
index 00000000..50807bac
--- /dev/null
+++ b/packages/core/test/runtime/session-layer-identity.test.ts
@@ -0,0 +1,526 @@
+/**
+ * Two things a session's identity has to get right, both of which the shared
+ * session log / warm-hydration optimisations quietly got wrong:
+ *
+ *  1. WHAT the shared log accepts. Every turn's context validates items against
+ *     the harness base registry EXTENDED with its context layers' `itemSchemas`,
+ *     but the session-owned log is created once per thread. Bind it to the base
+ *     registry and every item type a layer declares is rejected — on the seeding
+ *     path (`seedSessionHistory`, which the chat host calls on first contact for
+ *     every thread) and mid-turn alike. The seeding failure is caught and logged
+ *     upstream, so the symptom is not a crash but a thread that silently loses
+ *     its history on every message, forever.
+ *
+ *  2. WHICH bucket a warm turn inherits. Layer state lives under
+ *     `resolveScopeKey(layer.scope, ctx)`, and for 'resource' and 'global' scope
+ *     that key is not a function of the thread. Keying warm carry-forward on
+ *     threadId alone therefore hands one bucket's state to a turn addressing a
+ *     different one — and because write-through is then re-pointed at the NEW
+ *     scope key, the mismatch is persisted rather than staying in memory.
+ */
+
+import { describe, expect, it } from 'bun:test';
+import assert from 'node:assert';
+import type { ContextData, ContextLayer, StorageAdapter } from '@noetic-tools/context';
+import type { Item, LLMResponse, MessageItem, Step } from '@noetic-tools/types';
+import { frameworkCast } from '@noetic-tools/types';
+import { z } from 'zod';
+import { AgentHarness } from '../../src/harness/agent-harness';
+import { assistantMessage, makeStorage } from '../_helpers';
+
+//#region Helpers
+
+/** A custom item type contributed by a layer, not known to the base registry. */
+const NoteItemSchema = z.object({
+  type: z.literal('myapp:note'),
+  id: z.string(),
+  note: z.string(),
+});
+
+/**
+ * A custom item, shaped by the layer's own schema. `frameworkCast` is the seam a
+ * host crosses when handing its own declared item type to the framework: the
+ * registry is a shape GATE rather than a normalizer, so the item travels through
+ * the log as authored and `Item` cannot enumerate every extension type.
+ */
+function noteItem(id: string, note: string): Item {
+  return frameworkCast<Item>(
+    NoteItemSchema.parse({
+      type: 'myapp:note',
+      id,
+      note,
+    }),
+  );
+}
+
+/** An item whose type NO configured layer declares. */
+function undeclaredItem(): Item {
+  return frameworkCast<Item>({
+    type: 'myapp:unknown',
+    id: 'u1',
+  });
+}
+
+/** A layer that declares `myapp:note` as an item type. */
+function noteDeclaringLayer(): ContextLayer {
+  return {
+    id: 'notes',
+    slot: 100,
+    scope: 'thread',
+    itemSchemas: {
+      items: [
+        NoteItemSchema,
+      ],
+    },
+    hooks: {
+      async recall() {
+        return 'notes layer';
+      },
+    },
+  };
+}
+
+const chatStep: Step<ContextData, string, string> = {
+  kind: 'callModel',
+  id: 'chat',
+  model: 'test/scripted',
+  tools: [],
+};
+
+function harnessWith(opts: { layers?: ContextLayer[]; storage?: StorageAdapter }): AgentHarness {
+  let call = 0;
+  return new AgentHarness({
+    name: 'session-identity',
+    params: {},
+    agentGraph: chatStep,
+    environment: {
+      storage: {
+        adapter: opts.storage ?? makeStorage(),
+      },
+    },
+    contextLayers: opts.layers,
+    _testCallModel: async (): Promise<LLMResponse> => {
+      const message: MessageItem = assistantMessage(`answer ${call}`, `resp-${call}`);
+      call += 1;
+      return {
+        items: [
+          message,
+        ],
+        usage: {
+          inputTokens: 0,
+          outputTokens: 1,
+        },
+      };
+    },
+  });
+}
+
+interface SeenState {
+  seen: string[];
+}
+
+/**
+ * A layer at `scope` that records, in its own state, every event it saw — so a
+ * leak across scope boundaries is visible as one bucket's state containing
+ * another's marks. `tag` is whatever the current turn calls itself.
+ */
+function seenLayer(
+  scope: 'resource' | 'global' | 'thread',
+  tag: () => string,
+): {
+  layer: ContextLayer<SeenState>;
+  inits: () => number;
+} {
+  let inits = 0;
+  const layer: ContextLayer<SeenState> = {
+    id: 'seen',
+    slot: 100,
+    scope,
+    hooks: {
+      async init({ storage }) {
+        inits += 1;
+        const saved = await storage.get<SeenState>('state');
+        return {
+          state: saved ?? {
+            seen: [
+              `init@${tag()}`,
+            ],
+          },
+        };
+      },
+      async recall({ state }) {
+        return `seen=${state.seen.join(',')}`;
+      },
+      async store({ state }) {
+        return {
+          state: {
+            seen: [
+              ...(state?.seen ?? []),
+              `store@${tag()}`,
+            ],
+          },
+        };
+      },
+    },
+  };
+  return {
+    layer,
+    inits: () => inits,
+  };
+}
+
+async function runTurn(
+  harness: AgentHarness,
+  scope: {
+    threadId: string;
+    resourceId?: string;
+  },
+  text: string,
+): Promise<void> {
+  await harness.execute(text, scope);
+  await harness.getAgentResponse({
+    threadId: scope.threadId,
+  });
+}
+
+/** A layer's persisted state for one scope bucket, straight out of storage. */
+async function persistedState(
+  storage: StorageAdapter,
+  layerId: string,
+  scopeKey: string,
+): Promise<SeenState | null> {
+  return storage.get<SeenState>(`layers/${layerId}/${scopeKey}/state`);
+}
+
+//#endregion
+
+describe('the session-owned log honours the layers item registry', () => {
+  it('seedSessionHistory accepts an item type a context layer declares', async () => {
+    const harness = harnessWith({
+      layers: [
+        noteDeclaringLayer(),
+      ],
+    });
+    // Bound to the BASE registry this throws `item_schema_mismatch` from inside
+    // `seedSessionHistory` — swallowed by the caller upstream, so a thread's
+    // history vanishes silently on every message instead of failing loudly.
+    harness.seedSessionHistory('seeded', [
+      noteItem('n1', 'remembered'),
+    ]);
+
+    const items = await harness.previewRequestItems({
+      threadId: 'seeded',
+    });
+    expect(items.map((i) => i.type)).toContain('myapp:note');
+  });
+
+  it('a layer-declared item survives a full turn on the shared log', async () => {
+    const harness = harnessWith({
+      layers: [
+        noteDeclaringLayer(),
+      ],
+    });
+    harness.seedSessionHistory('turning', [
+      noteItem('n1', 'before the turn'),
+    ]);
+    await runTurn(
+      harness,
+      {
+        threadId: 'turning',
+      },
+      'hello',
+    );
+
+    const types = (
+      await harness.previewRequestItems({
+        threadId: 'turning',
+      })
+    ).map((i) => i.type);
+    // The custom item is still there, and the turn's own items landed alongside it.
+    expect(types).toContain('myapp:note');
+    expect(types).toContain('message');
+  });
+
+  it('an item type NO layer declares is still rejected', () => {
+    // The fix widens the log to the layers' registry — it must not widen it to
+    // anything, or `strictItemSchemas` would stop meaning anything on this path.
+    const harness = harnessWith({
+      layers: [
+        noteDeclaringLayer(),
+      ],
+    });
+    expect(() =>
+      harness.seedSessionHistory('rejecting', [
+        undeclaredItem(),
+      ]),
+    ).toThrow(/myapp:unknown/);
+  });
+
+  it('a harness with no layers still rejects an undeclared type', () => {
+    const harness = harnessWith({});
+    expect(() =>
+      harness.seedSessionHistory('bare', [
+        noteItem('n1', 'nobody declared me'),
+      ]),
+    ).toThrow(/myapp:note/);
+  });
+});
+
+describe('warm hydration respects each layers resolved scope key', () => {
+  it('a resource-scoped layer cold-inits when resourceId changes (no cross-tenant leak)', async () => {
+    /* The verified cross-tenant repro: one resource-scoped layer, one thread,
+     * turn 1 as 'alice' and turn 2 as 'bob'. Keyed on threadId alone, bob's turn
+     * took the warm path and inherited alice's state — then
+     * `registerDurableTargets` re-pointed write-through at scopeKey 'bob', so
+     * storage ended with bob's bucket holding alice's marks. */
+    const storage = makeStorage();
+    let tag = 'alice';
+    const probe = seenLayer('resource', () => tag);
+    const harness = harnessWith({
+      layers: [
+        probe.layer,
+      ],
+      storage,
+    });
+
+    await runTurn(
+      harness,
+      {
+        threadId: 'shared',
+        resourceId: 'alice',
+      },
+      'from alice',
+    );
+    tag = 'bob';
+    await runTurn(
+      harness,
+      {
+        threadId: 'shared',
+        resourceId: 'bob',
+      },
+      'from bob',
+    );
+
+    // Two buckets, each holding only its own marks.
+    const alice = await persistedState(storage, 'seen', 'alice');
+    const bob = await persistedState(storage, 'seen', 'bob');
+    assert(alice);
+    assert(bob);
+    expect(alice.seen).toEqual([
+      'init@alice',
+      'store@alice',
+    ]);
+    expect(bob.seen).toEqual([
+      'init@bob',
+      'store@bob',
+    ]);
+    // Explicitly: nothing of alice's reached bob's persisted record.
+    expect(bob.seen.some((s) => s.includes('alice'))).toBe(false);
+    // Bob's turn had to cold-init to read his own (empty) bucket.
+    expect(probe.inits()).toBe(2);
+  });
+
+  it('the same resourceId still warm-carries (the fix is not a blanket cold-init)', async () => {
+    const storage = makeStorage();
+    const probe = seenLayer('resource', () => 'alice');
+    const harness = harnessWith({
+      layers: [
+        probe.layer,
+      ],
+      storage,
+    });
+
+    for (const text of [
+      'one',
+      'two',
+      'three',
+    ]) {
+      await runTurn(
+        harness,
+        {
+          threadId: 'shared',
+          resourceId: 'alice',
+        },
+        text,
+      );
+    }
+
+    // One cold init for three turns — the whole point of the warm path.
+    expect(probe.inits()).toBe(1);
+    const alice = await persistedState(storage, 'seen', 'alice');
+    assert(alice);
+    // ...and state accumulated across them rather than resetting each turn.
+    expect(alice.seen.filter((s) => s.startsWith('store@'))).toHaveLength(3);
+  });
+
+  it('a resource-scoped layer returning to an earlier resourceId re-reads that bucket', async () => {
+    // alice → bob → alice. The final turn must resume ALICE's accumulated state,
+    // not bob's, and not a fresh init that discards hers.
+    const storage = makeStorage();
+    let tag = 'alice';
+    const probe = seenLayer('resource', () => tag);
+    const harness = harnessWith({
+      layers: [
+        probe.layer,
+      ],
+      storage,
+    });
+
+    await runTurn(
+      harness,
+      {
+        threadId: 'shared',
+        resourceId: 'alice',
+      },
+      'a1',
+    );
+    tag = 'bob';
+    await runTurn(
+      harness,
+      {
+        threadId: 'shared',
+        resourceId: 'bob',
+      },
+      'b1',
+    );
+    tag = 'alice';
+    await runTurn(
+      harness,
+      {
+        threadId: 'shared',
+        resourceId: 'alice',
+      },
+      'a2',
+    );
+
+    const alice = await persistedState(storage, 'seen', 'alice');
+    const bob = await persistedState(storage, 'seen', 'bob');
+    assert(alice);
+    assert(bob);
+    // Alice's two turns both stored; her bucket never saw bob.
+    expect(alice.seen.filter((s) => s === 'store@alice')).toHaveLength(2);
+    expect(alice.seen.some((s) => s.includes('bob'))).toBe(false);
+    expect(bob.seen.some((s) => s.includes('alice'))).toBe(false);
+  });
+
+  it('a global-scoped layer does not lose another threads update', async () => {
+    /* The verified lost-update repro: a global-scoped counter, thread A for three
+     * turns, thread B for one, then thread A again. A's warm entry held an
+     * in-memory copy predating B's increment, and the write-through mirror made
+     * that stale value win — turning a shared global bucket into a silent
+     * last-writer-wins race between threads. */
+    const storage = makeStorage();
+    let tag = 'A';
+    const probe = seenLayer('global', () => tag);
+    const harness = harnessWith({
+      layers: [
+        probe.layer,
+      ],
+      storage,
+    });
+
+    for (const text of [
+      'a1',
+      'a2',
+      'a3',
+    ]) {
+      await runTurn(
+        harness,
+        {
+          threadId: 'A',
+        },
+        text,
+      );
+    }
+    tag = 'B';
+    await runTurn(
+      harness,
+      {
+        threadId: 'B',
+      },
+      'b1',
+    );
+    tag = 'A';
+    await runTurn(
+      harness,
+      {
+        threadId: 'A',
+      },
+      'a4',
+    );
+
+    const global = await persistedState(storage, 'seen', '__global__');
+    assert(global);
+    // Five turns stored into the one shared bucket; none of them lost. The bug
+    // dropped B's, leaving four.
+    expect(global.seen.filter((s) => s.startsWith('store@'))).toHaveLength(5);
+    expect(global.seen).toContain('store@B');
+  });
+
+  it('a thread-scoped layer still warm-carries across turns (regression guard)', async () => {
+    // 'thread' is the scope where threadId happens to BE the scope key, so the
+    // pre-fix behaviour was already correct here. It must stay correct: the fix
+    // narrows carry-forward, and narrowing it too far would silently reintroduce
+    // a cold storage read per turn per layer.
+    const storage = makeStorage();
+    const probe = seenLayer('thread', () => 'T');
+    const harness = harnessWith({
+      layers: [
+        probe.layer,
+      ],
+      storage,
+    });
+
+    for (const text of [
+      'one',
+      'two',
+      'three',
+    ]) {
+      await runTurn(
+        harness,
+        {
+          threadId: 'T',
+        },
+        text,
+      );
+    }
+
+    expect(probe.inits()).toBe(1);
+    const state = await persistedState(storage, 'seen', 'T');
+    assert(state);
+    expect(state.seen.filter((s) => s.startsWith('store@'))).toHaveLength(3);
+  });
+
+  it('a resource-scoped layer with no resourceId falls back to the thread bucket', async () => {
+    // `resolveScopeKey('resource', ctx)` is `resourceId ?? threadId`, so turns
+    // without a resourceId all address the thread bucket and must warm-carry.
+    const storage = makeStorage();
+    const probe = seenLayer('resource', () => 'T');
+    const harness = harnessWith({
+      layers: [
+        probe.layer,
+      ],
+      storage,
+    });
+
+    await runTurn(
+      harness,
+      {
+        threadId: 'T',
+      },
+      'one',
+    );
+    await runTurn(
+      harness,
+      {
+        threadId: 'T',
+      },
+      'two',
+    );
+
+    expect(probe.inits()).toBe(1);
+    const state = await persistedState(storage, 'seen', 'T');
+    assert(state);
+    expect(state.seen.filter((s) => s.startsWith('store@'))).toHaveLength(2);
+  });
+});
diff --git a/packages/core/test/runtime/session-owned-log.test.ts b/packages/core/test/runtime/session-owned-log.test.ts
new file mode 100644
index 00000000..ffdada83
--- /dev/null
+++ b/packages/core/test/runtime/session-owned-log.test.ts
@@ -0,0 +1,133 @@
+/**
+ * The session-owned item log: one `ItemLogImpl` per thread, shared by reference
+ * with every turn's context. Two semantics the old copy-out/copy-back history
+ * gave for free must be preserved explicitly:
+ *
+ *  1. A FAILED turn leaves no trace. With a single shared log, the turn's
+ *     partial items (its input, anything appended before the failure) are
+ *     already in the log when the error hits — the runner rolls back to a
+ *     watermark captured at turn start instead of relying on never-copied-back.
+ *  2. History accumulates across turns by identity: turn N's context reads and
+ *     appends the same log turns 1..N-1 wrote, with no per-turn array spreads.
+ */
+
+import { describe, expect, it } from 'bun:test';
+import type { ContextData } from '@noetic-tools/context';
+import type { CallModelRequest, Item, LLMResponse, Step } from '@noetic-tools/types';
+import { AgentHarness } from '../../src/harness/agent-harness';
+import { textOnlyResponse } from '../_helpers';
+
+const echoStep: Step<ContextData, string, string> = {
+  kind: 'callModel',
+  id: 'echo',
+  model: 'test/echo',
+  tools: [],
+};
+
+function itemTexts(items: ReadonlyArray<Item>): string[] {
+  return items.flatMap((item) =>
+    item.type === 'message'
+      ? item.content.flatMap((part) =>
+          part.type === 'input_text' || part.type === 'output_text'
+            ? [
+                part.text,
+              ]
+            : [],
+        )
+      : [],
+  );
+}
+
+describe('session-owned item log', () => {
+  it('history accumulates across turns on one thread', async () => {
+    const captured: string[][] = [];
+    const harness = new AgentHarness({
+      name: 'test',
+      agentGraph: echoStep,
+      params: {},
+      _testCallModel: async (request: CallModelRequest): Promise<LLMResponse> => {
+        captured.push(itemTexts(request.items));
+        return textOnlyResponse(`reply ${captured.length}`);
+      },
+    });
+
+    await harness.execute('first');
+    await harness.getAgentResponse();
+    await harness.execute('second');
+    await harness.getAgentResponse();
+
+    expect(captured).toHaveLength(2);
+    // Turn 2 saw turn 1's input AND its reply — the shared log carried both
+    // forward without any copy-back step.
+    expect(captured[1]).toContain('first');
+    expect(captured[1]).toContain('reply 1');
+    expect(captured[1]).toContain('second');
+  });
+
+  it('a failed turn leaves no trace in the session history', async () => {
+    const captured: string[][] = [];
+    let fail = true;
+    const harness = new AgentHarness({
+      name: 'test',
+      agentGraph: echoStep,
+      params: {},
+      _testCallModel: async (request: CallModelRequest): Promise<LLMResponse> => {
+        captured.push(itemTexts(request.items));
+        if (fail) {
+          throw new Error('model exploded');
+        }
+        return textOnlyResponse('recovered');
+      },
+    });
+
+    await harness.execute('doomed turn');
+    await expect(harness.getAgentResponse()).rejects.toThrow('model exploded');
+
+    fail = false;
+    await harness.execute('next turn');
+    const response = await harness.getAgentResponse();
+    expect(response.text).toBe('recovered');
+
+    // The recovery turn's request contains neither the failed turn's input nor
+    // any partial output — the rollback restored the pre-turn watermark.
+    expect(captured).toHaveLength(2);
+    expect(captured[1]).toContain('next turn');
+    expect(captured[1]).not.toContain('doomed turn');
+  });
+
+  it('the failure rollback preserves earlier successful turns', async () => {
+    const captured: string[][] = [];
+    let fail = false;
+    const harness = new AgentHarness({
+      name: 'test',
+      agentGraph: echoStep,
+      params: {},
+      _testCallModel: async (request: CallModelRequest): Promise<LLMResponse> => {
+        captured.push(itemTexts(request.items));
+        if (fail) {
+          throw new Error('model exploded');
+        }
+        return textOnlyResponse(`reply ${captured.length}`);
+      },
+    });
+
+    await harness.execute('good turn');
+    await harness.getAgentResponse();
+
+    fail = true;
+    await harness.execute('bad turn');
+    await expect(harness.getAgentResponse()).rejects.toThrow('model exploded');
+
+    fail = false;
+    await harness.execute('final turn');
+    await harness.getAgentResponse();
+
+    const last = captured[captured.length - 1];
+    // Turn 1's exchange survived the failed turn's rollback...
+    expect(last).toContain('good turn');
+    expect(last).toContain('reply 1');
+    // ...while turn 2's input was erased.
+    expect(last).not.toContain('bad turn');
+    expect(last).toContain('final turn');
+  });
+});
diff --git a/specs/08-runtime.md b/specs/08-runtime.md
index aa92927e..14052c14 100644
--- a/specs/08-runtime.md
+++ b/specs/08-runtime.md
@@ -415,7 +415,9 @@ interface DetachedHandle<O> {
 
 - a FIFO `MessageQueue`,
 - a long-lived `EventBroadcaster` that relays SDK and framework events across all turns,
-- an `itemLog` snapshot that carries conversation history from turn to turn.
+- one live `ItemLogImpl` shared by every turn context in the session.
+
+The shared log eliminates per-turn copy-out/copy-back. A turn captures its starting length and truncates back to that watermark on failure or cancellation, so failed turns leave no partial input, model output, or tool results. History seeding validates the entire replacement before mutating the live log.
 
 Before the first turn runs, the harness walks the step tree to collect all tools from callModel steps, merges them with layer-provided tools, and deduplicates by name. This **unified tool set** is stored on the turn's execution context and sent with every LLM call for prompt cache efficiency. Individual steps restrict the model to a subset via the Open Responses `tool_choice: { type: "allowed_tools" }` parameter. `run(step, input, ctx)` does the same lazily: when an embedder drives a step directly on a bare `createContext()` context, `run` populates that context's unified tool set (the step's tools plus the harness tools) if it is not already set, so directly-driven steps — and any sub-agents they spawn — see the harness toolset. `run` also runs the configured context layers' `init()` hooks once per context (keyed by `ctx.id`) before executing, so a harness built with `contextLayers` + a storage adapter rehydrates prior layer state, recalls it, and persists updates on the bare `run()` path — the same guarantee the session/`execute()` path gives. The init is idempotent: nested or repeated `run()` calls and the session turn path never re-init (which would clobber accumulated in-layer state by re-hydrating from storage); a deliberate `disposeLayers(ctx)` clears the guard so a later `run()` re-hydrates.
 
diff --git a/specs/11-context-layer-system.md b/specs/11-context-layer-system.md
index 25d88761..438a174b 100644
--- a/specs/11-context-layer-system.md
+++ b/specs/11-context-layer-system.md
@@ -468,6 +468,10 @@ When `recall()` returns `tokenCount` less than allocated, the difference goes to
 
 The agent harness independently counts tokens. If layer-reported count diverges by >10%, the agent harness count is authoritative and a warning is emitted.
 
+## Warm Layer Hydration
+
+Session turns may carry initialized layer state forward without re-running `init`. Warm state is keyed by `(layer.id, resolveScopeKey(layer.scope, ctx))`, not by thread alone: resource-scoped state must cold-init when the resource changes, global state shares one bucket across threads, and thread state carries across turns. Execution-scoped layers always cold-init. Preview contexts are transient and never become a warm source. Each warm carry re-registers durable write-through targets for the new execution id.
+
 ## Recall Modes
 
 Each layer's `recallMode` controls whether its `recall()` blocks the model call: