agent-exec transport af2c39009af24615b572532ee5c44eaf/4f8f2959f25a

tests/test_task_recovery.py

@@ -43,6 +43,24 @@ class TaskRecoveryTests(unittest.TestCase):
         f.git(self.wt, "add", str(self.card.relative_to(self.wt)))
         f.git(self.wt, "commit", "-m", "Update fixture registration")
 
+    def local_authorization_scope(self):
+        """Capture the explicit registered in-progress task used as local-mode authorization.
+
+        The card carries the task scope (worktree path, branch, target branch, and R0
+        initial baseline) that authorize LOCAL_READY under a registered task, regardless
+        of whether the authoritative origin is currently reachable. This small synthetic
+        fixture exposes that scope as a single value so offline-local tests can name it."""
+        f = self.fixture
+        branch = f.git(self.wt, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
+        r0 = f.git(self.wt, "rev-parse", "origin/main").stdout.strip()
+        return {
+            "wt": self.wt,
+            "card": self.card,
+            "branch": branch,
+            "target_branch": "main",
+            "r0": r0,
+        }
+
     def test_local_precheck_allows_known_lag_without_changing_head(self):
         f = self.fixture
         head = f.git(self.wt, "rev-parse", "HEAD").stdout
@@ -88,17 +106,43 @@ class TaskRecoveryTests(unittest.TestCase):
         self.save_card_change()
         self.assertNotEqual(self.precheck().returncode, 0)
 
-    def test_local_precheck_rejects_dirty_state_and_unknown_remote(self):
+    def test_local_precheck_distinguishes_dirty_local_ready_and_remote_unverified(self):
+        """A registered in-progress task with explicit authorization and scope distinguishes
+        SHARED_WORKSPACE_BUSY (dirty), LOCAL_READY (offline, registered, HEAD unchanged),
+        and REMOTE_UNVERIFIED (delivery, origin unavailable)."""
         f = self.fixture
+        scope = self.local_authorization_scope()
+        self.assertEqual(scope["branch"], "feature/test")
+        self.assertTrue(scope["card"].is_file())
+        self.assertEqual(scope["target_branch"], "main")
+
+        # (1) Dirty workspace still yields SHARED_WORKSPACE_BUSY under the registered scope
+        # even when the authoritative origin is reachable. Clean before (2).
         (self.wt / "tracked").write_text("preserve current edit\n")
-        result = self.precheck()
-        self.assertNotEqual(result.returncode, 0)
-        self.assertIn("SHARED_WORKSPACE_BUSY", result.stdout)
+        dirty = self.precheck()
+        self.assertNotEqual(dirty.returncode, 0, dirty.stdout)
+        self.assertIn("[verdict] SHARED_WORKSPACE_BUSY", dirty.stdout)
+        self.assertNotIn("[verdict] LOCAL_READY", dirty.stdout)
         f.git(self.wt, "commit", "-am", "Save owner edit")
+
+        head_before = f.git(self.wt, "rev-parse", "HEAD").stdout
         f.git(self.wt, "remote", "set-url", "origin", str(f.root / "missing.git"))
-        result = self.precheck()
-        self.assertNotEqual(result.returncode, 0)
-        self.assertIn("REMOTE_UNVERIFIED", result.stdout)
+
+        # (2) Same unavailable origin; local mode against a registered in-progress task
+        # returns LOCAL_READY and leaves HEAD untouched.
+        local = self.precheck(mode="local")
+        self.assertEqual(local.returncode, 0, local.stdout)
+        self.assertIn("[verdict] LOCAL_READY", local.stdout)
+        self.assertNotIn("[verdict] REMOTE_UNVERIFIED", local.stdout)
+        self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
+
+        # (3) Same unavailable origin; delivery mode continues to refuse with
+        # REMOTE_UNVERIFIED because no read has refreshed the authoritative target.
+        delivery = self.precheck(mode="delivery")
+        self.assertNotEqual(delivery.returncode, 0, delivery.stdout)
+        self.assertIn("[verdict] REMOTE_UNVERIFIED", delivery.stdout)
+        self.assertNotIn("[verdict] LOCAL_READY", delivery.stdout)
+        self.assertEqual(f.git(self.wt, "rev-parse", "HEAD").stdout, head_before)
 
     def test_unknown_work_mode_is_rejected(self):
         result = self.precheck(mode="anything")