diff --git a/loopx/pi_goal_mode/README.md b/loopx/pi_goal_mode/README.md
index b68f43db8e..df7dce2a14 100644
--- a/loopx/pi_goal_mode/README.md
+++ b/loopx/pi_goal_mode/README.md
@@ -47,15 +47,31 @@ into a LoopX-governed visible goal loop.
   and fences an in-flight quota probe. For persistent sessions, the loop stays
   paused across Pi restarts until `/loopx resume` or a fresh goal activation
   explicitly re-arms it.
-- **Pause reasons** — every pause records why it happened in the binding:
-  `pauseReason: "user_prompt"` when a `before_agent_start` prompt was not the
-  loop's own injected task body, `"aborted"` after Escape, and `""` again once
-  `/loopx resume` re-arms the loop. Both pause paths report the pause in the UI
-  and the recorded reason is diagnostic only: it never decides continuation, and
-  the canonical `autoResume` flag stays the single authority. Prompt origin is not
-  inferred from another event: Pi compacts before it emits `before_agent_start`
-  for a submitted prompt, so a preceding `session_compact` does not prove that
-  the next prompt came from the host.
+- **Input boundary** — observed Pi `input` events with source `interactive` or
+  `rpc` pause automatic continuation, including queued `steer` / `followUp`,
+  image-only input and text identical to the last injected task. Pi still
+  processes the input and can compact/recover within that run; LoopX stops
+  submitting subsequent tasks. Already-submitted messages and running tools
+  are not cancelled. Input arriving during a pending resume or activation
+  revokes dispatch immediately, then waits for already-started re-arm writes
+  before saving the pause, so restarting Pi cannot restore that older re-arm.
+  Use `/loopx resume` to re-arm with fresh quota admission.
+- **Pause reasons** — a saved input pause records `pauseReason: "user_prompt"`,
+  Escape records `"aborted"`, and explicit resume clears the reason. Input and
+  run-start events for the same pause notify once and preserve an existing pause
+  reason. The reason is diagnostic; durable `autoResume` and LoopX quota still
+  authorize continuation. While saving an input pause, a session-local fence
+  also denies dispatch. If saving fails, that instance stays stopped and reports
+  that the pause may not survive restart; restore storage before explicit resume.
+- **Origin limits** — `extension` does not identify which extension sent input.
+  Unknown-source user starts retain the injected-text fallback; custom starts
+  pause because LoopX only injects user messages. The fallback compares the
+  injected text before waiting for a pending re-arm and persists a non-own
+  start's pause after older writes; a later explicit resume still takes priority.
+  Earlier Pi input handlers can consume an event before this extension sees it.
+  This is not complete prompt
+  provenance: compaction never exempts a prompt or re-arms a pause, and its
+  occurrence does not establish the next prompt's producer.
 
 ## Install / uninstall
 
diff --git a/loopx/pi_goal_mode/loopx-goal.ts b/loopx/pi_goal_mode/loopx-goal.ts
index 057365b5f5..e4c8b751ce 100644
--- a/loopx/pi_goal_mode/loopx-goal.ts
+++ b/loopx/pi_goal_mode/loopx-goal.ts
@@ -614,14 +614,30 @@ export default function (pi: ExtensionAPI) {
     await loop.interrupt(key);
   });
 
-  // Every prompt that is not this loop's own injected task body is the owner
-  // taking the wheel and pauses the loop; the pause itself carries the reason.
+  // Queued steer/followUp input does not start a new run. Observe its explicit
+  // host source here, even when its text equals our task body or is image-only.
+  // Return continue so Pi still processes the owner's message normally.
+  pi.on("input", async (event, ctx) => {
+    if (event.source === "interactive" || event.source === "rpc") {
+      const { key } = bindContext(ctx);
+      await loop.ownerInput(key);
+    }
+    return { action: "continue" };
+  });
+
+  // Starts without observed external input retain the conservative fallback.
+  // "extension" is not proof that a message came from this LoopX instance.
   // Pi compacts before it emits before_agent_start for a submitted prompt, so a
   // preceding compaction says nothing about who submitted that prompt and is not
   // used to exempt one.
   pi.on("before_agent_start", async (event, ctx) => {
     const { key } = bindContext(ctx);
-    await loop.userPrompt(key, String(event.prompt || ""));
+    if (event.messageType === "custom") {
+      // LoopX only injects user messages; a custom start has no prompt string.
+      await loop.ownerInput(key);
+    } else {
+      await loop.userPrompt(key, String(event.prompt || ""));
+    }
   });
 
   // Session shutdown / session replacement: atomically dispose the whole
diff --git a/loopx/pi_goal_mode/pi-goal-loop-runtime.mjs b/loopx/pi_goal_mode/pi-goal-loop-runtime.mjs
index 3b0d6e9ac1..c17ef05d75 100644
--- a/loopx/pi_goal_mode/pi-goal-loop-runtime.mjs
+++ b/loopx/pi_goal_mode/pi-goal-loop-runtime.mjs
@@ -810,9 +810,26 @@ export function createGoalLoop(options) {
   const evaluations = new Map()
   const contexts = new Map()
   const authorities = new Map()
+  // A local revocation fence also covers a pending/failed pause write. It can
+  // only deny continuation; durable binding + fresh quota still grant it.
+  const inputFences = new Map()
+  const pendingRearms = new Map()
   let disposed = false
   let epoch = 0
 
+  const writeRearm = (key, store, changes, expected) => {
+    const write = Promise.resolve().then(() => store.write(key, changes, expected))
+    // A newer input must persist its pause after every already-started re-arm,
+    // even if another activation/resume finishes before an older write. Failed
+    // re-arms do not prevent the input handler from trying to save its pause.
+    const pending = Promise.allSettled([pendingRearms.get(key), write])
+    pendingRearms.set(key, pending)
+    pending.then(() => {
+      if (pendingRearms.get(key) === pending) pendingRearms.delete(key)
+    })
+    return write
+  }
+
   const cancelScheduled = (key) => {
     const timer = timers.get(key)
     if (timer !== undefined) clearTimer(timer)
@@ -826,7 +843,7 @@ export function createGoalLoop(options) {
   }
 
   const scheduleEvaluation = (key, minutes) => {
-    if (disposed) return
+    if (disposed || inputFences.get(key)?.blocked) return
     cancelScheduled(key)
     const scheduledEpoch = epoch
     const timer = setTimer(async () => {
@@ -846,6 +863,10 @@ export function createGoalLoop(options) {
   const evaluateIdleOnce = async (key) => {
     if (disposed) return
     const instanceEpoch = epoch
+    const inputFence = inputFences.get(key)
+    const isCurrent = () => !disposed && epoch === instanceEpoch &&
+      inputFences.get(key) === inputFence && !inputFence?.blocked
+    if (!isCurrent()) return
     const services = contexts.get(key)
     if (!services) {
       cancelScheduled(key)
@@ -861,10 +882,11 @@ export function createGoalLoop(options) {
     try {
       binding = await store.read(key)
     } catch {
+      if (!isCurrent()) return
       cancelScheduled(key)
       return
     }
-    if (disposed || epoch !== instanceEpoch) return
+    if (!isCurrent()) return
     if (!binding || binding.terminal) {
       cancelScheduled(key)
       return
@@ -885,15 +907,15 @@ export function createGoalLoop(options) {
     try {
       decision = await quotaProbe(binding)
     } catch {
-      if (disposed || epoch !== instanceEpoch) return
+      if (!isCurrent()) return
       // Fail closed: never continue without LoopX authority. Bounded retry.
       scheduleEvaluation(key, DEFAULT_RETRY_MINUTES)
       return
     }
-    if (disposed || epoch !== instanceEpoch) return
+    if (!isCurrent()) return
 
     const current = await store.read(key)
-    if (disposed || epoch !== instanceEpoch) return
+    if (!isCurrent()) return
     if (
       !current ||
       current.terminal ||
@@ -908,7 +930,7 @@ export function createGoalLoop(options) {
     const applyDecision = async (nextDecision, allowFinalCheck = true) => {
       if (nextDecision?.scheduler_hint?.action === "stop_until_explicit_resume" && !isTerminalNoFollowup(nextDecision)) {
         const committed = await store.write(key, { autoResume: false }, expected)
-        if (!committed || disposed || epoch !== instanceEpoch) return
+        if (!committed || !isCurrent()) return
         cancelScheduled(key)
         services.notify(`LoopX goal ${current.goalId} paused; resolve its Core stop condition, then run /loopx resume.`, "info")
         return
@@ -918,11 +940,11 @@ export function createGoalLoop(options) {
         // activated a new goal while this write was in-flight, the commit is
         // rejected and the new binding stays alive.
         const committed = await store.write(key, { terminal: true, autoResume: false }, expected)
+        if (!isCurrent()) return
         if (!committed) {
           cancelScheduled(key)
           return
         }
-        if (disposed || epoch !== instanceEpoch) return
         cancelScheduled(key)
         services.notify(
           `LoopX goal ${current.goalId} reached validated terminal no-follow-up; loop stopped.`,
@@ -938,18 +960,18 @@ export function createGoalLoop(options) {
           { schedulerToken: "", unchangedPolls: 0 },
           expected,
         )
+        if (!isCurrent()) return
         if (!schedulerCommit) {
           cancelScheduled(key)
           return
         }
-        if (disposed || epoch !== instanceEpoch) return
         const prompt = current.taskBody || current.goalId
         const promptCommit = await store.write(key, { lastInjectedPrompt: prompt }, expected)
+        if (!isCurrent()) return
         if (!promptCommit) {
           cancelScheduled(key)
           return
         }
-        if (disposed || epoch !== instanceEpoch) return
         sendMessage(prompt)
         return
       }
@@ -962,13 +984,13 @@ export function createGoalLoop(options) {
         try {
           finalDecision = await quotaProbe(current)
         } catch {
-          if (disposed || epoch !== instanceEpoch) return
+          if (!isCurrent()) return
           scheduleEvaluation(key, DEFAULT_RETRY_MINUTES)
           return
         }
-        if (disposed || epoch !== instanceEpoch) return
+        if (!isCurrent()) return
         const latest = await store.read(key)
-        if (disposed || epoch !== instanceEpoch) return
+        if (!isCurrent()) return
         if (
           !latest ||
           latest.terminal ||
@@ -992,11 +1014,11 @@ export function createGoalLoop(options) {
         },
         expected,
       )
+      if (!isCurrent()) return
       if (!waitCommit) {
         cancelScheduled(key)
         return
       }
-      if (disposed || epoch !== instanceEpoch) return
       scheduleEvaluation(key, wait.minutes)
     }
 
@@ -1004,13 +1026,14 @@ export function createGoalLoop(options) {
     } catch {
       // Fail closed with bounded retry: any unhandled error (store write
       // failure, etc.) must not break the evaluation chain permanently.
+      if (!isCurrent()) return
       cancelScheduled(key)
       scheduleEvaluation(key, DEFAULT_RETRY_MINUTES)
     }
   }
 
   const evaluateIdle = (key) => {
-    if (disposed) return Promise.resolve()
+    if (disposed || inputFences.get(key)?.blocked) return Promise.resolve()
     const existing = evaluations.get(key)
     if (existing) return existing
     const evaluation = evaluateIdleOnce(key).finally(() => {
@@ -1020,6 +1043,62 @@ export function createGoalLoop(options) {
     return evaluation
   }
 
+  // Undefined prompt means the adapter observed external input, independently
+  // of its text. A string retains the legacy before_agent_start fallback.
+  const pauseForInput = async (key, prompt) => {
+    if (disposed) return
+    const instanceEpoch = epoch
+    const services = contexts.get(key)
+    if (!services) return
+    let fence = inputFences.get(key)
+    const isCurrent = () => !disposed && epoch === instanceEpoch && inputFences.get(key) === fence
+    const revoke = () => {
+      fence = { blocked: true }
+      inputFences.set(key, fence)
+      cancelScheduled(key)
+    }
+    // Revoke synchronously, before storage can yield to an in-flight quota
+    // evaluation. Pi catches input-handler errors, so throwing is not a gate.
+    if (prompt === undefined) revoke()
+    try {
+      if (prompt !== undefined) {
+        // Classify before waiting for a re-arm: its old paused binding can
+        // still identify our own injection. Every other start must revoke the
+        // older operation, even when its write has not reached disk yet.
+        const binding = await services.store.read(key)
+        if (!isCurrent()) return
+        if (binding && prompt === binding.lastInjectedPrompt) return
+        revoke()
+      }
+      // A paused snapshot is not final while an older resume/activation can
+      // still save autoResume=true. Revoke dispatch now, then read after those
+      // writes finish so this pause also survives a new runtime instance.
+      await pendingRearms.get(key)
+      if (!isCurrent()) return
+      const binding = await services.store.read(key)
+      if (!isCurrent()) return
+      if (!binding || binding.terminal || binding.autoResume === false) return
+      const committed = await services.store.write(key, {
+        autoResume: false,
+        generation: (binding.generation || 0) + 1,
+        pauseReason: "user_prompt",
+      }, { generation: binding.generation || 0, goalId: binding.goalId, autoResume: true })
+      if (!isCurrent()) return
+      if (!committed) throw new Error("binding changed while saving the pause")
+      services.notify(
+        `LoopX goal ${committed.goalId} auto-continuation paused by user input; run /loopx resume to continue.`,
+        "info",
+      )
+    } catch {
+      if (!isCurrent()) return
+      revoke()
+      services.notify(
+        "LoopX auto-continuation stopped for this session, but the input pause could not be saved. Restore state storage, then run /loopx resume; the pause may not survive a restart.",
+        "error",
+      )
+    }
+  }
+
   return {
     // Bind the ctx-derived services (store, isIdle, notify) for a session key.
     // Re-binding on every event keeps the loop free of host types while still
@@ -1062,11 +1141,15 @@ export function createGoalLoop(options) {
       if (!services) throw new Error("loopx_goal_activate requires a bound session context")
       const authority = authorities.get(key) || services.authority
       const verified = assertActivationAuthority(authority, fields)
+      const inputFence = { blocked: true }
+      inputFences.set(key, inputFence)
+      cancelScheduled(key)
       // Increment the persisted generation so every in-flight evaluation for
       // this key is rejected at its next compare-and-swap commit.
       const current = await services.store.read(key)
+      if (disposed || inputFences.get(key) !== inputFence) return null
       const generation = (current?.generation || 0) + 1
-      const binding = await services.store.write(key, {
+      const binding = await writeRearm(key, services.store, {
         ...fields,
         activationToken: verified.token,
         goalId: verified.goalId,
@@ -1079,6 +1162,10 @@ export function createGoalLoop(options) {
         throw authorityError("authority_mismatch", "Pi session binding changed during activation")
       }
       if (disposed) return null
+      if (inputFences.get(key) !== inputFence) return null
+      // Completion is still the same re-arm. A fallback classifying its text
+      // must distinguish completion from a later explicit resume/activation.
+      inputFence.blocked = false
       cancelScheduled(key)
       services.notify(
         `LoopX goal ${binding.goalId} activated; continuation gated by LoopX quota.`,
@@ -1101,44 +1188,26 @@ export function createGoalLoop(options) {
       if (!binding || binding.terminal || binding.autoResume === false) return
       await evaluateIdle(key)
     },
-    // Every prompt that is not this loop's own injected task body is treated as
-    // the owner taking the wheel, and the pause records why it happened. The
-    // loop's own turns are recognized through lastInjectedPrompt; no other event
-    // is used to infer a prompt's origin, because Pi compacts before it emits
-    // before_agent_start for a submitted prompt, so a preceding session_compact
-    // says nothing about who submitted that prompt.
-    async userPrompt(key, prompt) {
-      if (disposed) return
-      const instanceEpoch = epoch
-      const services = contexts.get(key)
-      if (!services) return
-      let binding = null
-      try {
-        binding = await services.store.read(key)
-      } catch {
-        return
-      }
-      if (disposed || epoch !== instanceEpoch) return
-      if (!binding || binding.terminal) return
-      if (String(prompt || "") !== binding.lastInjectedPrompt) {
-        const expected = { generation: binding.generation || 0, goalId: binding.goalId }
-        const committed = await services.store.write(
-          key,
-          {
-            autoResume: false,
-            generation: (binding.generation || 0) + 1,
-            pauseReason: "user_prompt",
-          },
-          expected,
-        )
-        if (!committed) return
-        if (disposed || epoch !== instanceEpoch) return
-        cancelScheduled(key)
-        services.notify(
-          `LoopX goal ${committed.goalId} auto-continuation paused by user input; run /loopx resume to continue.`,
-          "info",
-        )
+    // Host-observed interactive/RPC input always revokes automatic dispatch,
+    // including queued, image-only and text-identical input. This does not
+    // abort Pi, consume its input or remove messages already handed to Pi.
+    ownerInput(key) {
+      const pending = inputFences.get(key)?.pending
+      if (pending) return pending
+      const pause = pauseForInput(key)
+      const fence = inputFences.get(key)
+      if (fence?.blocked) {
+        // Concurrent RPC input shares the same pause commit and notification.
+        fence.pending = pause
+        const clearPending = () => { delete fence.pending }
+        pause.then(clearPending, clearPending)
       }
+      return pause
+    },
+    // Unknown-source starts retain the text fallback; compaction is never
+    // evidence of origin. Neither this fallback nor input can re-arm a pause.
+    userPrompt(key, prompt) {
+      return pauseForInput(key, String(prompt || ""))
     },
     async interrupt(key) {
       if (disposed) return null
@@ -1171,9 +1240,14 @@ export function createGoalLoop(options) {
       if (disposed) return null
       const services = contexts.get(key)
       if (!services) return null
+      // Separate this explicit re-arm from an older pending input pause. A new
+      // input must revoke this attempt, not coalesce into that older pause.
+      const inputFence = { blocked: true }
+      inputFences.set(key, inputFence)
+      cancelScheduled(key)
       const current = await services.store.read(key)
-      if (!current || disposed) return null
-      const binding = await services.store.write(key,
+      if (!current || disposed || inputFences.get(key) !== inputFence) return null
+      const binding = await writeRearm(key, services.store,
         {
           autoResume: true,
           terminal: false,
@@ -1183,6 +1257,8 @@ export function createGoalLoop(options) {
         { generation: current.generation || 0, goalId: current.goalId })
       if (!binding) return null
       if (disposed) return null
+      if (inputFences.get(key) !== inputFence) return null
+      inputFence.blocked = false
       cancelScheduled(key)
       services.notify(`LoopX goal ${binding.goalId} auto-continuation resumed.`, "info")
       return binding
diff --git a/tests/pi_goal_adapter_input.test.mjs b/tests/pi_goal_adapter_input.test.mjs
new file mode 100644
index 0000000000..c318e67203
--- /dev/null
+++ b/tests/pi_goal_adapter_input.test.mjs
@@ -0,0 +1,240 @@
+import assert from "node:assert/strict";
+import { promises as fs } from "node:fs";
+import { mkdtemp, rm } from "node:fs/promises";
+import { createRequire, registerHooks, syncBuiltinESMExports } from "node:module";
+import os from "node:os";
+import path from "node:path";
+import test from "node:test";
+import { createBindingStore, sessionKey } from "../loopx/pi_goal_mode/pi-goal-loop-runtime.mjs";
+
+// Actual adapter and durable binding; SDK/UI and CLI admission are synthetic.
+// Pi's input contract delivers queued steer/followUp without before_agent_start.
+registerHooks({
+  resolve(specifier, context, next) {
+    if (["typebox", "@earendil-works/pi-ai", "@earendil-works/pi-tui"].includes(specifier))
+      return { url: "pi-input-test:sdk", shortCircuit: true };
+    return next(specifier, context);
+  },
+  load(url, context, next) {
+    if (url === "pi-input-test:sdk") return {
+      format: "module", shortCircuit: true,
+      source: "export const Type = new Proxy({}, {get: () => () => ({})}); export const StringEnum = () => ({}); export class Text {}",
+    };
+    return next(url, context);
+  },
+});
+let cliReply;
+const childProcess = createRequire(import.meta.url)("node:child_process");
+childProcess.execFile = Object.assign(() => {}, {
+  [Symbol.for("nodejs.util.promisify.custom")]: async (_binary, args) => ({
+    stdout: JSON.stringify(cliReply(args)),
+  }),
+});
+syncBuiltinESMExports();
+const { default: extension } = await import("../loopx/pi_goal_mode/loopx-goal.ts");
+
+async function harness(t, { prime = true } = {}) {
+  const directory = await mkdtemp(path.join(os.tmpdir(), "pi-owner-input-"));
+  const file = path.join(directory, "session.jsonl");
+  const commands = new Map(), tools = new Map(), events = new Map();
+  const outgoing = [], notices = [];
+  let idle = true, probes = 0;
+  cliReply = args => {
+    if (args.includes("should-run")) {
+      probes += 1;
+      return { ok: true, should_run: true };
+    }
+    return {
+      goal_id: "synthetic-goal", project: "synthetic-project",
+      project_connection: { registry: "synthetic-registry", connection_state: "connected" },
+      host_loop_activation: {
+        activation_allowed: true, agent_id: "synthetic-agent",
+        available_capabilities: ["task_lease_v0"],
+      },
+    };
+  };
+  const pi = {
+    registerCommand: (name, command) => commands.set(name, command),
+    registerTool: tool => tools.set(tool.name, tool),
+    registerEntryRenderer() {}, appendEntry() {},
+    on: (name, handler) => events.set(name, handler),
+    sendUserMessage: text => outgoing.push(text),
+  };
+  extension(pi);
+  const ctx = {
+    cwd: directory, isIdle: () => idle,
+    sessionManager: { getSessionFile: () => file },
+    ui: { setWidget() {}, notify: (message, kind) => notices.push({message, kind}) },
+  };
+  const emit = (name, event = {}) => events.get(name)?.({type: name, ...event}, ctx);
+  const command = text => commands.get("loopx").handler(text, ctx);
+  const store = createBindingStore(directory);
+  const read = () => store.read(sessionKey(file));
+  const settle = async () => { idle = true; await emit("agent_settled"); };
+  t.after(async () => {
+    await emit("session_shutdown");
+    await rm(directory, {recursive: true, force: true});
+  });
+  await command("Continue the synthetic task");
+  const packet = JSON.parse(outgoing.shift());
+  const result = await tools.get("loopx_goal_activate").execute("test", {
+    activationToken: packet.pi_session_authority.token, objective: "Synthetic task body",
+  }, undefined, undefined, ctx);
+  assert.equal(JSON.parse(result.content[0].text).ok, true);
+  if (prime) {
+    await settle();
+    const prompt = outgoing.shift();
+    assert.equal(prompt, "Synthetic task body");
+    await emit("input", {source: "extension", text: prompt});
+    await emit("before_agent_start", {messageType: "user", prompt});
+    assert.equal((await read()).autoResume, true);
+  }
+  notices.length = 0;
+  return {directory, emit, command, read, settle, outgoing, notices,
+    restart: async () => { await emit("session_shutdown"); extension(pi); },
+    busy: () => { idle = false; }, probes: () => probes};
+}
+
+for (const source of ["interactive", "rpc"]) {
+  for (const streamingBehavior of ["steer", "followUp"]) {
+    test(source + " " + streamingBehavior + " pauses subsequent dispatch across compaction", async t => {
+      const h = await harness(t);
+      h.busy();
+      const before = await h.read();
+      const input = {source, streamingBehavior, text: "Inspect the synthetic result"};
+      assert.deepEqual(await h.emit("input", input), {action: "continue"});
+      const paused = await h.read();
+      assert.equal(paused.autoResume, false);
+      assert.equal(paused.generation, before.generation + 1);
+      assert.equal(paused.pauseReason, "user_prompt");
+      assert.equal(h.notices.length, 1);
+      // Recovery inside the current run does not re-arm automatic dispatch.
+      await h.emit("session_compact", {reason: "overflow", willRetry: true});
+      await h.emit("agent_end", {messages: [{role: "assistant", stopReason: "stop"}]});
+      const probes = h.probes();
+      await h.settle();
+      await h.settle();
+      assert.equal(h.probes(), probes);
+      assert.equal(h.outgoing.length, 0);
+      await h.command("resume");
+      await h.settle();
+      assert.equal(h.probes(), probes + 1);
+      assert.equal(h.outgoing.length, 1);
+    });
+  }
+
+  test(source + " input identical to the injected body pauses exactly once", async t => {
+    const h = await harness(t);
+    const before = await h.read();
+    await h.emit("input", {source, text: before.lastInjectedPrompt});
+    assert.equal((await h.read()).autoResume, false);
+    await h.emit("session_compact", {reason: "threshold", willRetry: false});
+    await h.emit("before_agent_start", {messageType: "user", prompt: before.lastInjectedPrompt});
+    assert.equal((await h.read()).autoResume, false);
+    await h.emit("before_agent_start", {messageType: "user", prompt: "Expanded input"});
+    const paused = await h.read();
+    assert.equal(paused.autoResume, false);
+    assert.equal(paused.generation, before.generation + 1);
+    assert.equal(h.notices.length, 1);
+    await h.settle();
+    assert.equal(h.outgoing.length, 0);
+  });
+}
+
+test("image-only input before the first injection pauses", async t => {
+  const h = await harness(t, {prime: false});
+  const images = [{type: "image", mimeType: "image/png", data: "synthetic"}];
+  await h.emit("input", {source: "interactive", text: "", images});
+  await h.emit("before_agent_start", {messageType: "user", prompt: "", images});
+  assert.equal((await h.read()).autoResume, false);
+  assert.equal(h.notices.length, 1);
+  await h.settle();
+  assert.equal(h.outgoing.length, 0);
+});
+
+test("own extension input continues; foreign extension starts retain the fallback", async t => {
+  const h = await harness(t);
+  await h.emit("session_compact", {reason: "manual", willRetry: false});
+  await h.settle();
+  assert.equal(h.outgoing.length, 1);
+  assert.equal((await h.read()).autoResume, true);
+  await h.emit("input", {source: "extension", text: "Foreign extension task"});
+  await h.emit("before_agent_start", {messageType: "user", prompt: "Foreign extension task"});
+  assert.equal((await h.read()).autoResume, false);
+});
+
+test("custom starts cannot match an empty lastInjectedPrompt", async t => {
+  const h = await harness(t, {prime: false});
+  await h.emit("before_agent_start", {
+    messageType: "custom",
+    message: {role: "custom", customType: "synthetic", content: "", display: false, timestamp: 0},
+  });
+  assert.equal((await h.read()).autoResume, false);
+  await h.settle();
+  assert.equal(h.outgoing.length, 0);
+});
+
+test("an input observed after abort preserves the existing pause reason", async t => {
+  const h = await harness(t);
+  await h.emit("agent_end", {messages: [{role: "assistant", stopReason: "aborted"}]});
+  const aborted = await h.read();
+  await h.emit("input", {source: "interactive", text: "Inspect the result"});
+  await h.emit("before_agent_start", {messageType: "user", prompt: "Inspect the result"});
+  assert.equal((await h.read()).pauseReason, "aborted");
+  assert.equal((await h.read()).generation, aborted.generation);
+});
+
+for (const kind of ["foreign extension", "own extension", "interactive", "rpc"]) {
+  test(kind + " start during a pending resume survives adapter restart", async t => {
+    const h = await harness(t);
+    h.busy();
+    await h.emit("agent_end", {messages: [{role: "assistant", stopReason: "aborted"}]});
+    const before = await h.read();
+    assert.equal(before.autoResume, false);
+    const entered = Promise.withResolvers(), release = Promise.withResolvers();
+    const rename = fs.rename;
+    let blocked = false;
+    // Delay the real atomic file commit, without replacing the store or CAS.
+    t.mock.method(fs, "rename", async (from, to) => {
+      if (!blocked && String(to).startsWith(h.directory + path.sep)) {
+        const value = JSON.parse(await fs.readFile(from, "utf8"));
+        if (value.autoResume) {
+          blocked = true;
+          entered.resolve();
+          await release.promise;
+        }
+      }
+      return rename(from, to);
+    });
+    t.after(() => release.resolve());
+    const resume = h.command("resume");
+    await entered.promise;
+    const own = kind === "own extension";
+    const source = kind.endsWith("extension") ? "extension" : kind;
+    const text = own ? before.lastInjectedPrompt : "Inspect the new result before continuing";
+    const input = h.emit("input", {source, text});
+    // Explicit input finishes before Pi starts its run; the extension input
+    // handler continues immediately and only its start supplies the fallback.
+    const start = source === "extension"
+      ? h.emit("before_agent_start", {messageType: "user", prompt: text})
+      : input.then(() => h.emit("before_agent_start", {messageType: "user", prompt: text}));
+    release.resolve();
+    await Promise.all([resume, input, start]);
+    assert.equal((await h.read()).autoResume, own);
+    assert.equal((await h.read()).pauseReason, own ? "" : "user_prompt");
+    for (const restart of [false, true]) {
+      if (restart) await h.restart();
+      const probes = h.probes(), messages = h.outgoing.length;
+      await h.settle();
+      assert.equal(h.probes() - probes, Number(own));
+      assert.equal(h.outgoing.length - messages, Number(own));
+    }
+    if (!own) {
+      const probes = h.probes(), messages = h.outgoing.length;
+      await h.command("resume");
+      await h.settle();
+      assert.equal(h.probes() - probes, 1, "later explicit resume obtains fresh admission");
+      assert.equal(h.outgoing.length - messages, 1);
+    }
+  });
+}
diff --git a/tests/pi_goal_loop_runtime.test.mjs b/tests/pi_goal_loop_runtime.test.mjs
index aa8f32a0bc..47b1819767 100644
--- a/tests/pi_goal_loop_runtime.test.mjs
+++ b/tests/pi_goal_loop_runtime.test.mjs
@@ -1741,3 +1741,481 @@ test("file-backed Core pause survives a new loop instance and resume rechecks ad
     second.loop.dispose()
   } finally { await fs.rm(root, { recursive: true, force: true }) }
 })
+
+
+test("explicit owner input ignores text equality and duplicate starts do not re-pause", async () => {
+  const f = harness({ should_run: true })
+  f.loop.bind("owner", f.services)
+  await f.activate("owner")
+  await f.loop.settle("owner")
+  const before = await f.store.read("owner")
+  await Promise.all([f.loop.ownerInput("owner"), f.loop.ownerInput("owner")])
+  await f.loop.ownerInput("owner")
+  await f.loop.userPrompt("owner", before.lastInjectedPrompt)
+  await f.loop.userPrompt("owner", "expanded input")
+  const paused = await f.store.read("owner")
+  assert.equal(paused.autoResume, false)
+  assert.equal(paused.generation, before.generation + 1)
+  assert.equal(paused.pauseReason, "user_prompt")
+  assert.equal(f.calls.notify, 2) // activation + one pause
+  await f.loop.settle("owner")
+  assert.equal(f.calls.send, 1)
+})
+
+for (const failure of ["read", "write", "cas"]) {
+  test("input pause " + failure + " failure keeps dispatch revoked until successful resume", async () => {
+    const f = harness(backoffDecision())
+    const notices = []
+    const inner = f.store
+    let fail = false, failResume = false
+    const store = {
+      read: key => {
+        if (fail && failure === "read") throw new Error("synthetic read failure")
+        return inner.read(key)
+      },
+      write: (key, changes, expected) => {
+        if (fail && changes.autoResume === false) {
+          if (failure === "write") throw new Error("synthetic write failure")
+          if (failure === "cas") return null
+        }
+        if (failResume && changes.autoResume === true) throw new Error("synthetic resume failure")
+        return inner.write(key, changes, expected)
+      },
+    }
+    f.loop.bind("fault", {...f.services, store, notify: (message, kind) => notices.push({message, kind})})
+    await f.activate("fault")
+    await f.loop.settle("fault")
+    const timer = f.scheduled.at(-1)
+    fail = true
+    await f.loop.ownerInput("fault")
+    fail = false
+    assert.equal((await inner.read("fault")).autoResume, true, "persistence did not succeed")
+    assert.equal(notices.at(-1).kind, "error")
+    assert.match(notices.at(-1).message, /could not be saved/)
+    assert.equal(timer.cleared, true)
+    f.setDecision({ should_run: true })
+    await timer.callback()
+    await f.loop.settle("fault")
+    assert.equal(f.calls.quota, 1)
+    assert.equal(f.calls.send, 0)
+    failResume = true
+    await assert.rejects(f.loop.resume("fault"), /resume failure/)
+    await f.loop.settle("fault")
+    assert.equal(f.calls.send, 0)
+    failResume = false
+    await f.loop.resume("fault")
+    await f.loop.settle("fault")
+    assert.equal(f.calls.quota, 2)
+    assert.equal(f.calls.send, 1)
+  })
+}
+
+test("owner input fences a committed dispatch while its pause cannot be persisted", async () => {
+  const f = harness({ should_run: true })
+  const entered = Promise.withResolvers(), release = Promise.withResolvers()
+  let failPause = false
+  const store = {
+    read: key => f.store.read(key),
+    async write(key, changes, expected) {
+      if (failPause && changes.autoResume === false) throw new Error("synthetic pause failure")
+      const result = await f.store.write(key, changes, expected)
+      if (changes.lastInjectedPrompt) {
+        entered.resolve()
+        await release.promise
+      }
+      return result
+    },
+  }
+  f.loop.bind("dispatch", {...f.services, store})
+  await f.activate("dispatch")
+  const pending = f.loop.settle("dispatch")
+  await entered.promise
+  failPause = true
+  await f.loop.ownerInput("dispatch")
+  release.resolve()
+  await pending
+  assert.equal(f.calls.send, 0)
+  assert.equal(f.scheduled.length, 0)
+})
+
+test("input revokes synchronously while its state read is pending", async () => {
+  const probe = Promise.withResolvers(), probeStarted = Promise.withResolvers()
+  const readStarted = Promise.withResolvers(), releaseRead = Promise.withResolvers()
+  const f = harness(() => { probeStarted.resolve(); return probe.promise })
+  let blockRead = false
+  const store = {
+    async read(key) {
+      if (blockRead) {
+        blockRead = false
+        readStarted.resolve()
+        await releaseRead.promise
+      }
+      return f.store.read(key)
+    },
+    write: (...args) => f.store.write(...args),
+  }
+  f.loop.bind("pending", {...f.services, store})
+  await f.activate("pending")
+  const evaluation = f.loop.settle("pending")
+  await probeStarted.promise
+  blockRead = true
+  const pause = f.loop.ownerInput("pending")
+  await readStarted.promise
+  probe.resolve({ should_run: true })
+  await evaluation
+  assert.equal(f.calls.send, 0)
+  releaseRead.resolve()
+  await pause
+  assert.equal((await f.store.read("pending")).autoResume, false)
+})
+
+test("a newer activation fences a delayed owner-input read", async () => {
+  const entered = Promise.withResolvers(), release = Promise.withResolvers()
+  const f = harness({ should_run: true })
+  let blockRead = false
+  const store = {
+    async read(key) {
+      if (blockRead) {
+        blockRead = false
+        entered.resolve()
+        await release.promise
+      }
+      return f.store.read(key)
+    },
+    write: (...args) => f.store.write(...args),
+  }
+  f.loop.bind("replacement", {...f.services, store})
+  await f.activate("replacement")
+  blockRead = true
+  const pause = f.loop.ownerInput("replacement")
+  await entered.promise
+  await f.activate("replacement", {taskBody: "new task"})
+  release.resolve()
+  await pause
+  await f.loop.settle("replacement")
+  assert.equal((await f.store.read("replacement")).autoResume, true)
+  assert.deepEqual(f.calls.messages, ["new task"])
+})
+
+test("input revocation is scoped to its session and survives a runtime restart on disk", async t => {
+  const root = await fs.mkdtemp(path.join(os.tmpdir(), "pi-input-pause-"))
+  t.after(() => fs.rm(root, {recursive: true, force: true}))
+  const store = createBindingStore(root)
+  const f = harness({ should_run: true })
+  for (const key of ["a", "b"]) {
+    f.loop.bind(key, {...f.services, store})
+    await f.activate(key)
+  }
+  await f.loop.ownerInput("a")
+  await f.loop.settle("a")
+  await f.loop.settle("b")
+  assert.equal(f.calls.send, 1)
+  f.loop.dispose()
+  const restarted = harness({ should_run: true })
+  restarted.loop.bind("a", {...restarted.services, store: createBindingStore(root)})
+  await restarted.loop.settle("a")
+  assert.equal(restarted.calls.quota, 0)
+  assert.equal(restarted.calls.send, 0)
+  await restarted.loop.resume("a")
+  await restarted.loop.settle("a")
+  assert.equal(restarted.calls.quota, 1)
+  assert.equal(restarted.calls.send, 1)
+  restarted.loop.dispose()
+})
+
+test("new input during resume does not coalesce into the pre-resume pause", async () => {
+  const f = harness({ should_run: true })
+  const firstPause = Promise.withResolvers(), releasePause = Promise.withResolvers()
+  const resuming = Promise.withResolvers(), releaseResume = Promise.withResolvers()
+  let gatePause = true, gateResume = false
+  const store = {
+    read: key => f.store.read(key),
+    async write(key, changes, expected) {
+      if (changes.autoResume === false && gatePause) {
+        gatePause = false
+        firstPause.resolve()
+        await releasePause.promise
+      }
+      if (changes.autoResume === true && gateResume) {
+        resuming.resolve()
+        await releaseResume.promise
+      }
+      return f.store.write(key, changes, expected)
+    },
+  }
+  f.loop.bind("overlap", {...f.services, store})
+  await f.activate("overlap")
+  const oldPause = f.loop.ownerInput("overlap")
+  await firstPause.promise
+  gateResume = true
+  const resume = f.loop.resume("overlap")
+  await resuming.promise
+  const newPause = f.loop.ownerInput("overlap")
+  releaseResume.resolve()
+  await resume
+  releasePause.resolve()
+  await Promise.all([oldPause, newPause])
+  assert.equal((await f.store.read("overlap")).autoResume, false)
+  await f.loop.settle("overlap")
+  assert.equal(f.calls.send, 0)
+})
+
+for (const [operation, inputKind] of [
+  ["resume", "explicit"], ["activation", "explicit"],
+  ["resume", "fallback"], ["activation", "fallback"],
+]) {
+  for (const phase of ["read", "write"]) {
+    test(`${inputKind} input during a pending ${operation} ${phase} stays paused after restart`, async t => {
+      const root = await fs.mkdtemp(path.join(os.tmpdir(), "pi-rearm-input-"))
+      t.after(() => fs.rm(root, { recursive: true, force: true }))
+      const disk = createBindingStore(root)
+      const f = harness({ should_run: true })
+      t.after(() => f.loop.dispose())
+      f.loop.bind("durable-race", { ...f.services, store: disk })
+      await f.activate("durable-race", { autoResume: false, pauseReason: "aborted" })
+      const paused = await disk.read("durable-race")
+      const entered = Promise.withResolvers(), release = Promise.withResolvers()
+      let blockRead = phase === "read", blockWrite = phase === "write"
+      let pendingWrite = false
+      const store = {
+        async read(key) {
+          if (blockRead) {
+            blockRead = false
+            entered.resolve()
+            await release.promise
+            return paused
+          }
+          // The blocked write has not changed the file. Return its actual
+          // paused snapshot synchronously to fix this ordering without timers.
+          if (pendingWrite) return paused
+          return disk.read(key)
+        },
+        async write(key, changes, expected) {
+          if (blockWrite && changes.autoResume === true) {
+            blockWrite = false
+            pendingWrite = true
+            entered.resolve()
+            await release.promise
+            pendingWrite = false
+          }
+          return disk.write(key, changes, expected)
+        },
+      }
+      f.loop.bind("durable-race", { ...f.services, store })
+      const rearm = operation === "resume"
+        ? f.loop.resume("durable-race") : f.activate("durable-race")
+      await entered.promise
+      const pause = () => inputKind === "explicit"
+        ? f.loop.ownerInput("durable-race")
+        : f.loop.userPrompt("durable-race", "Inspect the new result")
+      const input = pause()
+      const duplicate = pause()
+      await f.loop.settle("durable-race")
+      assert.equal(f.calls.send, 0, "input revokes dispatch before the older operation finishes")
+      release.resolve()
+      assert.equal(await rearm, null, "the older operation cannot report a successful re-arm")
+      await Promise.all([input, duplicate])
+      const saved = await disk.read("durable-race")
+      assert.equal(saved.autoResume, false, "newer input must be persisted after the older re-arm")
+      assert.equal(saved.pauseReason, phase === "read" ? "aborted" : "user_prompt")
+      await f.loop.ownerInput("durable-race")
+      assert.equal((await disk.read("durable-race")).generation, saved.generation)
+      f.loop.dispose()
+
+      const restarted = harness({ should_run: true })
+      t.after(() => restarted.loop.dispose())
+      restarted.loop.bind("durable-race", { ...restarted.services, store: createBindingStore(root) })
+      await restarted.loop.settle("durable-race")
+      assert.equal(restarted.calls.quota, 0)
+      assert.equal(restarted.calls.send, 0)
+      await restarted.loop.resume("durable-race")
+      await restarted.loop.settle("durable-race")
+      assert.equal(restarted.calls.quota, 1, "a later explicit resume obtains fresh admission")
+      assert.equal(restarted.calls.send, 1)
+    })
+  }
+}
+
+test("input waits for an older activation even after a newer re-arm finishes", async () => {
+  const f = harness({ should_run: true })
+  f.loop.bind("multiple", f.services)
+  await f.activate("multiple", { autoResume: false })
+  const entered = Promise.withResolvers(), release = Promise.withResolvers()
+  let block = true
+  const store = {
+    read: key => f.store.read(key),
+    async write(key, changes, expected) {
+      if (block && changes.autoResume === true) {
+        block = false
+        entered.resolve()
+        await release.promise
+      }
+      return f.store.write(key, changes, expected)
+    },
+  }
+  f.loop.bind("multiple", { ...f.services, store })
+  const older = f.activate("multiple")
+  await entered.promise
+  await f.activate("multiple")
+  const pause = f.loop.ownerInput("multiple")
+  await f.loop.settle("multiple")
+  assert.equal(f.calls.send, 0)
+  release.resolve()
+  await Promise.all([older, pause])
+  assert.equal((await f.store.read("multiple")).autoResume, false)
+  await f.loop.settle("multiple")
+  assert.equal(f.calls.quota, 0)
+  f.loop.dispose()
+})
+
+test("a later explicit resume supersedes input waiting for an older resume", async () => {
+  const f = harness({ should_run: true })
+  f.loop.bind("later", f.services)
+  await f.activate("later", { autoResume: false })
+  const entered = Promise.withResolvers(), release = Promise.withResolvers()
+  let block = true
+  const store = {
+    read: key => f.store.read(key),
+    async write(key, changes, expected) {
+      if (block && changes.autoResume === true) {
+        block = false
+        entered.resolve()
+        await release.promise
+      }
+      return f.store.write(key, changes, expected)
+    },
+  }
+  f.loop.bind("later", { ...f.services, store })
+  const older = f.loop.resume("later")
+  await entered.promise
+  const pause = f.loop.ownerInput("later")
+  assert.ok(await f.loop.resume("later"))
+  release.resolve()
+  await Promise.all([older, pause])
+  assert.equal((await f.store.read("later")).autoResume, true)
+  await f.loop.settle("later")
+  assert.equal(f.calls.quota, 1)
+  assert.equal(f.calls.send, 1)
+  f.loop.dispose()
+})
+
+test("a failed pending resume does not prevent input from retaining a saved pause", async () => {
+  const f = harness({ should_run: true })
+  f.loop.bind("failed-rearm", f.services)
+  await f.activate("failed-rearm", { autoResume: false, pauseReason: "aborted" })
+  const entered = Promise.withResolvers(), release = Promise.withResolvers()
+  const store = {
+    read: key => f.store.read(key),
+    async write(key, changes, expected) {
+      if (changes.autoResume === true) {
+        entered.resolve()
+        await release.promise
+        throw new Error("synthetic resume failure")
+      }
+      return f.store.write(key, changes, expected)
+    },
+  }
+  f.loop.bind("failed-rearm", { ...f.services, store })
+  const failed = assert.rejects(f.loop.resume("failed-rearm"), /synthetic resume failure/)
+  await entered.promise
+  const pause = f.loop.ownerInput("failed-rearm")
+  release.resolve()
+  await Promise.all([failed, pause])
+  assert.equal((await f.store.read("failed-rearm")).autoResume, false)
+  assert.equal((await f.store.read("failed-rearm")).pauseReason, "aborted")
+  assert.equal(f.calls.notify, 1, "no false unsaved-pause warning or resume notification")
+  f.loop.bind("failed-rearm", f.services)
+  await f.loop.resume("failed-rearm")
+  await f.loop.settle("failed-rearm")
+  assert.equal(f.calls.send, 1)
+  f.loop.dispose()
+})
+
+for (const laterResume of [false, true]) {
+  test(`fallback classification spanning a completed resume honors later resume=${laterResume}`, async t => {
+    const f = harness({ should_run: true })
+    t.after(() => f.loop.dispose())
+    f.loop.bind("classify", f.services)
+    await f.activate("classify", { autoResume: false })
+    const writing = Promise.withResolvers(), releaseWrite = Promise.withResolvers()
+    const reading = Promise.withResolvers(), releaseRead = Promise.withResolvers()
+    let blockWrite = true, blockRead = false
+    const store = {
+      async read(key) {
+        const value = await f.store.read(key)
+        if (blockRead) {
+          blockRead = false
+          reading.resolve()
+          await releaseRead.promise
+        }
+        return value
+      },
+      async write(key, changes, expected) {
+        if (blockWrite && changes.autoResume === true) {
+          blockWrite = false
+          writing.resolve()
+          await releaseWrite.promise
+        }
+        return f.store.write(key, changes, expected)
+      },
+    }
+    f.loop.bind("classify", { ...f.services, store })
+    const resume = f.loop.resume("classify")
+    await writing.promise
+    blockRead = true
+    const input = f.loop.userPrompt("classify", "Inspect the new result")
+    // Let the older resume finish while classification can still be pending.
+    releaseWrite.resolve()
+    await resume
+    // A dropped input may finish without ever attempting classification.
+    await Promise.race([reading.promise, input])
+    blockRead = false
+    if (laterResume) await f.loop.resume("classify")
+    releaseRead.resolve()
+    await input
+    assert.equal((await f.store.read("classify")).autoResume, laterResume)
+    await f.loop.settle("classify")
+    assert.equal(f.calls.quota, Number(laterResume))
+    assert.equal(f.calls.send, Number(laterResume))
+  })
+}
+
+test("failed fallback classification keeps a pending resume locally revoked", async t => {
+  const f = harness({ should_run: true })
+  t.after(() => f.loop.dispose())
+  f.loop.bind("read-failure", f.services)
+  await f.activate("read-failure", { autoResume: false })
+  const entered = Promise.withResolvers(), release = Promise.withResolvers()
+  let failRead = false, blockWrite = true
+  const notices = []
+  const store = {
+    read(key) {
+      if (failRead) throw new Error("synthetic classification failure")
+      return f.store.read(key)
+    },
+    async write(key, changes, expected) {
+      if (blockWrite && changes.autoResume === true) {
+        blockWrite = false
+        entered.resolve()
+        await release.promise
+      }
+      return f.store.write(key, changes, expected)
+    },
+  }
+  f.loop.bind("read-failure", { ...f.services, store, notify: (...args) => notices.push(args) })
+  const resume = f.loop.resume("read-failure")
+  await entered.promise
+  failRead = true
+  const input = f.loop.userPrompt("read-failure", "Inspect the new result")
+  release.resolve()
+  await Promise.all([resume, input])
+  failRead = false
+  assert.ok(notices.some(([text, kind]) => kind === "error" && /could not be saved/.test(text)))
+  await f.loop.settle("read-failure")
+  assert.equal(f.calls.quota, 0)
+  assert.equal(f.calls.send, 0)
+  await f.loop.resume("read-failure")
+  await f.loop.settle("read-failure")
+  assert.equal(f.calls.quota, 1)
+  assert.equal(f.calls.send, 1)
+})
diff --git a/tests/test_pi_goal_mode.py b/tests/test_pi_goal_mode.py
index 5a63d7ff2f..0f71714242 100644
--- a/tests/test_pi_goal_mode.py
+++ b/tests/test_pi_goal_mode.py
@@ -218,3 +218,10 @@ def test_pi_goal_adapter_authority_contract() -> None:
     assert node, "Node.js >=22.22.3 is required for the shipped Pi adapter"
     test_file = Path(__file__).with_name("pi_goal_adapter_authority.test.mjs")
     subprocess.run([node, "--test", str(test_file)], check=True)
+
+
+def test_pi_goal_adapter_input_contract() -> None:
+    node = shutil.which("node")
+    assert node, "Node.js >=22.22.3 is required for the shipped Pi adapter"
+    test_file = Path(__file__).with_name("pi_goal_adapter_input.test.mjs")
+    subprocess.run([node, "--test", str(test_file)], check=True)