Phase 5.22: task_archive helper + bridge archive wiring (sync OSS sutando)

src/discord-bridge.py

@@ -57,6 +57,7 @@
 from single_instance import acquire as _single_instance_acquire  # noqa: E402
 from util_paths import shared_personal_path  # noqa: E402
 from task_priority import default_priority_for_source  # noqa: E402
+from task_archive import find_task_file  # noqa: E402
 REPO = resolve_workspace()
 
 # Lazy import for cloud telemetry — guarded so a missing cloud_metrics.py
@@ -2900,7 +2901,7 @@ async def poll_results():
                     # post — same UX as [no-send] / [REPLIED].
                     print(f"  Skipped (already replied or deduped): {task_id}")
                     archive_file(result_file, "results", task_id)
-                    task_file = TASKS_DIR / f"{task_id}.txt"
+                    task_file = find_task_file(TASKS_DIR, task_id) or TASKS_DIR / f"{task_id}.txt"
                     archive_file(task_file, "tasks", task_id)
                     continue
                 try:
@@ -3056,7 +3057,7 @@ async def poll_results():
                     print(f"  Reply failed: {e}", flush=True)
                 # Archive (not delete) so we can mine patterns later.
                 archive_file(result_file, "results", task_id)
-                task_file = TASKS_DIR / f"{task_id}.txt"
+                task_file = find_task_file(TASKS_DIR, task_id) or TASKS_DIR / f"{task_id}.txt"
                 archive_file(task_file, "tasks", task_id)
         await asyncio.sleep(1)
 
@@ -3264,8 +3265,8 @@ async def poll_dm_fallback():
                     # Archive matching task file so audit_orphan_tasks sees
                     # the task as processed (even if drop-without-reply).
                     _task_id = f.stem
-                    _task_file = TASKS_DIR / f"{_task_id}.txt"
-                    if _task_file.exists():
+                    _task_file = find_task_file(TASKS_DIR, _task_id)
+                    if _task_file:
                         archive_file(_task_file, "tasks", _task_id)
                     continue
                 # Stop retrying after 24h. Without this cap, a permanent
@@ -3277,8 +3278,8 @@ async def poll_dm_fallback():
                     print(f"  [dm-fallback] dropping stale {f.name} (age={int(age)}s)", flush=True)
                     f.unlink(missing_ok=True)
                     _task_id = f.stem
-                    _task_file = TASKS_DIR / f"{_task_id}.txt"
-                    if _task_file.exists():
+                    _task_file = find_task_file(TASKS_DIR, _task_id)
+                    if _task_file:
                         archive_file(_task_file, "tasks", _task_id)
                     continue
                 # Honor result-body suppression markers (parity with the
@@ -3293,8 +3294,8 @@ async def poll_dm_fallback():
                 if _peek.startswith('[no-send]') or _peek.startswith('[REPLIED]') or _peek.startswith('[deduped:'):
                     print(f"  [dm-fallback] skipped (suppression marker): {f.name}", flush=True)
                     _task_id = f.stem
-                    _task_file = TASKS_DIR / f"{_task_id}.txt"
-                    if _task_file.exists():
+                    _task_file = find_task_file(TASKS_DIR, _task_id)
+                    if _task_file:
                         archive_file(_task_file, "tasks", _task_id)
                     archive_file(f, "results", _task_id)
                     continue
@@ -3374,8 +3375,8 @@ async def poll_dm_fallback():
                                     # send. Log only.
                                     print(f"  [dm-fallback channel-redirect] file marker, file not found: {fpath}", file=sys.stderr, flush=True)
                             print(f"  [dm-fallback channel-redirect] sent {f.name} to channel {target_channel_id}", flush=True)
-                            _task_file = TASKS_DIR / f"{_task_id}.txt"
-                            if _task_file.exists():
+                            _task_file = find_task_file(TASKS_DIR, _task_id)
+                            if _task_file:
                                 archive_file(_task_file, "tasks", _task_id)
                             archive_file(f, "results", _task_id)
                             continue
@@ -3445,8 +3446,8 @@ async def poll_dm_fallback():
                     except OSError:
                         _result_text = ""
                     archive_file(f, "results", _task_id)
-                    _task_file = TASKS_DIR / f"{_task_id}.txt"
-                    if _task_file.exists():
+                    _task_file = find_task_file(TASKS_DIR, _task_id)
+                    if _task_file:
                         archive_file(_task_file, "tasks", _task_id)
                     if _result_text and _task_id.startswith("task-"):
                         # urlopen is blocking — run in thread so we don't stall

src/slack-bridge.py

@@ -47,6 +47,7 @@
 from task_priority import default_priority_for_source  # noqa: E402
 from result_markers import parse_markers  # noqa: E402
 from workspace_default import resolve_workspace  # noqa: E402
+from task_archive import find_task_file  # noqa: E402
 from single_instance import acquire as _single_instance_acquire  # noqa: E402
 
 try:
@@ -667,7 +668,7 @@ def result_watcher():
                         print(f"[Slack] reply error: {e}", flush=True)
 
                 archive_file(result_file, "results", task_id)
-                archive_file(TASKS_DIR / f"{task_id}.txt", "tasks", task_id)
+                archive_file(find_task_file(TASKS_DIR, task_id) or TASKS_DIR / f"{task_id}.txt", "tasks", task_id)
 
             # Proactive messages (sent to owner DM)
             if not presenter_mode_active():

src/task_archive.py

@@ -0,0 +1,34 @@
+"""Task-file locator for archive calls (#933).
+
+claim_task.py (#884) renames task-{id}.txt → task-{id}.claimed-core-N.txt
+when a core claims work. Bridge archive calls that hard-code the bare
+task-{id}.txt path silently no-op after claiming, leaving stranded
+.claimed-core-N.txt files in tasks/ forever.
+
+Usage:
+    from task_archive import find_task_file
+
+    task_file = find_task_file(TASKS_DIR, task_id)
+    if task_file:
+        archive_file(task_file, "tasks", task_id)
+
+Ported verbatim from OSS sutando in Phase 5.22 of the OSS → private sync.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def find_task_file(tasks_dir: Path, task_id: str) -> Path | None:
+    """Return the actual task file path for task_id, or None if absent.
+
+    Checks the bare name first (unclaimed), then globs for the claimed
+    variant (task-{id}.claimed-core-N.txt). If multiple claimed variants
+    exist (shouldn't happen but defensive), returns the first lexicographic
+    match and that's good enough — the caller only needs one path to archive.
+    """
+    bare = tasks_dir / f"{task_id}.txt"
+    if bare.exists():
+        return bare
+    matches = sorted(tasks_dir.glob(f"{task_id}.claimed-core-*.txt"))
+    return matches[0] if matches else None

src/telegram-bridge.py

@@ -29,6 +29,7 @@ def _push_vision_image(path: str, source: str = "telegram") -> bool:  # type: ig
 from result_markers import parse_markers  # noqa: E402
 
 from workspace_default import resolve_workspace  # noqa: E402
+from task_archive import find_task_file  # noqa: E402
 from single_instance import acquire as _single_instance_acquire  # noqa: E402
 REPO = resolve_workspace()
 TASKS_DIR = REPO / "tasks"
@@ -752,7 +753,7 @@ def main():
                 if any(a.kind == "skip" for a in parsed.actions):
                     print(f"  Skipped (marker): {task_id}", flush=True)
                     archive_file(result_file, "results", task_id)
-                    task_file = TASKS_DIR / f"{task_id}.txt"
+                    task_file = find_task_file(TASKS_DIR, task_id) or TASKS_DIR / f"{task_id}.txt"
                     archive_file(task_file, "tasks", task_id)
                     continue
                 try:
@@ -762,7 +763,7 @@ def main():
                     print(f"[Telegram] Reply error: {e}", flush=True)
                 # Archive (not delete) so we can mine patterns later.
                 archive_file(result_file, "results", task_id)
-                task_file = TASKS_DIR / f"{task_id}.txt"
+                task_file = find_task_file(TASKS_DIR, task_id) or TASKS_DIR / f"{task_id}.txt"
                 archive_file(task_file, "tasks", task_id)
 
         time.sleep(1)

tests/state-paths-adoption.test.py

@@ -159,6 +159,11 @@
     # runtime-state.
     "src/util_paths.py",
     "src/util_paths.ts",
+    # task_archive.py is a pure locator helper — it takes tasks_dir as a
+    # parameter from the caller and never resolves workspace itself. The
+    # flagged token appears only in the module docstring (example usage),
+    # not in runnable code.
+    "src/task_archive.py",
     # core_heartbeat is intentionally dep-free per its own comment —
     # must run before any other Sutando module is loaded, so it inlines
     # the workspace resolution rather than importing workspace_default.

tests/task-archive.test.py

@@ -0,0 +1,55 @@
+"""Tests for src/task_archive.py — find_task_file() helper (closes #933)."""
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+import sys
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
+from task_archive import find_task_file
+
+
+class TestFindTaskFile(unittest.TestCase):
+    def setUp(self) -> None:
+        self._td = tempfile.TemporaryDirectory()
+        self.tasks_dir = Path(self._td.name)
+        self.addCleanup(self._td.cleanup)
+
+    def _write(self, name: str, content: str = "task body") -> Path:
+        p = self.tasks_dir / name
+        p.write_text(content)
+        return p
+
+    def test_bare_file_returned(self) -> None:
+        self._write("task-123.txt")
+        result = find_task_file(self.tasks_dir, "task-123")
+        self.assertIsNotNone(result)
+        self.assertEqual(result.name, "task-123.txt")
+
+    def test_claimed_file_returned_when_bare_missing(self) -> None:
+        self._write("task-456.claimed-core-2.txt")
+        result = find_task_file(self.tasks_dir, "task-456")
+        self.assertIsNotNone(result)
+        self.assertEqual(result.name, "task-456.claimed-core-2.txt")
+
+    def test_bare_preferred_over_claimed(self) -> None:
+        self._write("task-789.txt")
+        self._write("task-789.claimed-core-1.txt")
+        result = find_task_file(self.tasks_dir, "task-789")
+        self.assertEqual(result.name, "task-789.txt")
+
+    def test_returns_none_when_no_file(self) -> None:
+        result = find_task_file(self.tasks_dir, "task-nonexistent")
+        self.assertIsNone(result)
+
+    def test_multiple_claimed_returns_first_lexicographic(self) -> None:
+        self._write("task-000.claimed-core-2.txt")
+        self._write("task-000.claimed-core-3.txt")
+        result = find_task_file(self.tasks_dir, "task-000")
+        self.assertIsNotNone(result)
+        self.assertIn("claimed-core-", result.name)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)