fix(cron,skills): tolerate non-UTF-8 cron archives and org markers

agent/skill_utils.py

@@ -44,7 +44,7 @@ def read_active_org_id(skills_dir: Path) -> Optional[str]:
     marker = skills_dir / ORG_MIRROR_DIR_NAME / ORG_ACTIVE_MARKER
     try:
         return (marker.read_text(encoding="utf-8").strip() or None) if marker.exists() else None
-    except OSError:
+    except (OSError, UnicodeDecodeError):
         return None
 
 

cron/scheduler_prompt.py

@@ -122,7 +122,14 @@ def _inject_context_from(job: dict, prompt: str) -> tuple[str, bool]:
             )
             latest_output = ""
             for output_file in output_files:
-                candidate = output_file.read_text(encoding="utf-8").strip()
+                try:
+                    candidate = output_file.read_text(encoding="utf-8").strip()
+                except (OSError, UnicodeDecodeError) as e:
+                    # Unreadable or non-UTF-8 archive: skip it like a silent/blank
+                    # file so an older usable archive still gets used.
+                    logger.warning(
+                        "context_from: skipping unreadable archive %s: %s", output_file, e)
+                    continue
                 # Only the run header describes suppression; script/agent payloads can
                 # quote these markers. Keep error documents useful for recovery context.
                 header = candidate.split("\n---\n", 1)[0].split("\n## Prompt", 1)[0]

tests/agent/test_org_skill_namespace.py

@@ -72,6 +72,18 @@ def test_helpers(self, tmp_path):
         _mark_active(skills, "org-9")
         assert sku.read_active_org_id(skills) == "org-9"
 
+    def test_undecodable_marker_treated_as_no_org(self, tmp_path):
+        """A non-UTF-8 .active_org marker resolves as no org instead of raising."""
+        skills = tmp_path / "skills"
+        org_root = skills / sku.ORG_MIRROR_DIR_NAME
+        org_root.mkdir(parents=True)
+        (org_root / sku.ORG_ACTIVE_MARKER).write_bytes(b"\xff\xfe\x00corrupt")
+        assert sku.read_active_org_id(skills) is None
+        # The gate call inside iter_skill_index_files must not raise either.
+        _mk_skill(skills, "personal-a")
+        found = [p.parent.name for p in sku.iter_skill_index_files(skills, "SKILL.md")]
+        assert "personal-a" in found
+
 
 class TestSnapshotEntryProvenance:
     def test_org_entry_strips_prefix_and_carries_provenance(self, tmp_path):

tests/cron/test_context_from_response_extraction.py

@@ -98,3 +98,64 @@ def test_headingless_archive_injects_whole_document(self, cron_env):
         assert injected is True
         assert "plain script payload" in prompt
         assert "line two" in prompt
+
+
+class TestUnreadableArchives:
+    """A corrupt archive file must be skipped like a silent one, not kill the run."""
+
+    def test_corrupt_newest_archive_falls_through_to_older(self, cron_env):
+        """Newest .md holds non-UTF-8 bytes: skip it and use the older good one."""
+        import os
+        from cron.jobs import OUTPUT_DIR, create_job
+        from cron.scheduler_prompt import _inject_context_from
+
+        source = create_job(prompt="Upstream", schedule="0 8 * * *")
+        _write_archive(
+            cron_env, source["id"], "2026-09-18_08-00-00.md",
+            "# Cron Job: up\n\n## Prompt\n\ngo\n\n## Response\n\nOLDER-GOOD\n",
+        )
+        corrupt = OUTPUT_DIR / source["id"] / "2026-09-19_08-00-00.md"
+        corrupt.write_bytes(b"\xff\xfe\x00corrupt")
+        os.utime(corrupt, (2_000_000_000, 2_000_000_000))
+
+        job = create_job(prompt="Report", schedule="0 8 * * *")
+        job["context_from"] = [source["id"]]
+
+        prompt, injected = _inject_context_from(job, "Report")
+
+        assert injected is True
+        assert "OLDER-GOOD" in prompt
+
+    def test_corrupt_only_archive_skips_source(self, cron_env):
+        """A source whose only archive is undecodable is skipped, not fatal."""
+        from cron.jobs import OUTPUT_DIR, create_job
+        from cron.scheduler_prompt import _inject_context_from
+
+        source = create_job(prompt="Upstream", schedule="0 8 * * *")
+        out_dir = OUTPUT_DIR / source["id"]
+        out_dir.mkdir(parents=True)
+        (out_dir / "2026-09-19_08-00-00.md").write_bytes(b"\xff\xfe\x00corrupt")
+
+        job = create_job(prompt="Report", schedule="0 8 * * *")
+        job["context_from"] = [source["id"]]
+
+        assert _inject_context_from(job, "Report") == ("Report", False)
+
+    def test_build_job_prompt_survives_corrupt_archive(self, cron_env):
+        """E2E through the real run path: _build_job_prompt must not raise on a
+        corrupt archive - the job fires with no injected context."""
+        from cron.jobs import OUTPUT_DIR, create_job
+        from cron.scheduler_prompt import _build_job_prompt
+
+        source = create_job(prompt="Upstream", schedule="0 8 * * *")
+        out_dir = OUTPUT_DIR / source["id"]
+        out_dir.mkdir(parents=True)
+        (out_dir / "2026-09-19_08-00-00.md").write_bytes(b"\xff\xfe\x00corrupt")
+
+        job = create_job(prompt="Report", schedule="0 8 * * *")
+        job["context_from"] = [source["id"]]
+
+        prompt = _build_job_prompt(job)
+
+        assert "Report" in prompt
+        assert "Output from job" not in prompt