Recover research pipelines after registry removal

docs/implementation-checkpoints.md

@@ -18,6 +18,8 @@ The remaining human gates are external sandbox submission, external disclosure d
 
 Follow-up: bridge service installation now persists the selected OpenCodex provider/model identifier in the fixed service command. This prevents an autonomous background worker from silently reverting to a provider default after the browser closes or the workstation restarts. The service file still contains no provider credentials.
 
+Live acceptance exposed a retracted-package recovery gap: an exact NuGet version previously collected and hashed had disappeared from the registry. Investigation pipelines now prefer a fresh official-registry collection but may reuse an exact prior quarantine artifact after independently verifying ecosystem, package, version, byte size, SHA-256, regular-file state, and configured size limit. Missing or altered cache entries still fail closed. The recorded step makes registry unavailability and quarantine reuse explicit, and no package code is executed.
+
 ## 107 Hermes Agent v1.0.0 Integration
 
 Status: complete in production.

docs/research-automation.md

@@ -18,6 +18,8 @@ The normal workflow no longer requires an evidence-bundle export, a file upload,
 
 If the bridge or collection step fails, click **Retry from checkpoint**. A new pipeline revision is created, stale proposals are superseded, and the previous revision remains auditable. If comparison was incomplete, enter a verified reference and click **Add reference and rerun analysis**.
 
+If an exact package version is later removed from its registry, the pipeline may reuse a previously collected local quarantine artifact only when its ecosystem, package, version, byte size, and SHA-256 all match. The reuse reason and failed registry retrieval are recorded in the step result. SecOpsAI never substitutes another version or trusts an unverified local file.
+
 The pipeline never executes package code, submits an artifact to an external sandbox, sends external communication, approves publication, or publishes an article. In agent-review mode it may record a bounded case verdict, but only against accepted, pipeline-specific evidence and with all guardrail decisions retained in the audit trail.
 
 ## Agent-review mode

secopsai/research_intake.py

@@ -602,6 +602,35 @@ def collect_package_intake(
     }
 
 
+def validate_quarantined_intake(result: Dict[str, Any]) -> Dict[str, Any]:
+    """Verify that a normalized intake still matches its owner-only quarantined artifact."""
+    metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {}
+    analysis = result.get("analysis") if isinstance(result.get("analysis"), dict) else {}
+    quarantine = result.get("quarantine") if isinstance(result.get("quarantine"), dict) else {}
+    digest = _text(metadata.get("artifact_sha256") or quarantine.get("artifact_id"), 64).lower()
+    if not re.fullmatch(r"[a-f0-9]{64}", digest):
+        raise IntakeError("cached intake does not contain a valid artifact hash")
+    filename = _text(analysis.get("filename") or "artifact", 512)
+    path = _quarantine_path(digest, filename)
+    if not path.is_file() or path.is_symlink():
+        raise IntakeError("cached quarantined artifact is unavailable")
+    if path.stat().st_size > MAX_ARTIFACT_BYTES:
+        raise IntakeError("cached quarantined artifact exceeds the configured size limit")
+    actual = hashlib.sha256(path.read_bytes()).hexdigest()
+    if actual != digest:
+        raise IntakeError("cached quarantined artifact hash changed")
+    expected_bytes = int(metadata.get("artifact_bytes") or quarantine.get("bytes") or 0)
+    if expected_bytes and expected_bytes != path.stat().st_size:
+        raise IntakeError("cached quarantined artifact size changed")
+    return {
+        "artifact_sha256": digest,
+        "artifact_bytes": path.stat().st_size,
+        "filename": filename,
+        "verified": True,
+        "execution_performed": False,
+    }
+
+
 def run_package_intake(
     *,
     case_id: str,

secopsai/research_pipeline.py

@@ -13,7 +13,12 @@
 from secopsai.intelligence_jobs import cancel_job, enqueue_job, get_job
 from secopsai.research_analysis import compare_intakes
 from secopsai.research_cases import add_evidence, get_case
-from secopsai.research_intake import attach_intake_result, collect_package_intake
+from secopsai.research_intake import (
+    IntakeError,
+    attach_intake_result,
+    collect_package_intake,
+    validate_quarantined_intake,
+)
 from secopsai.research_workflow import (
     build_evidence_matrix,
     publication_safety_check,
@@ -401,7 +406,12 @@ def _run_pipeline(pipeline_id: str, *, actor: str, db_path: Optional[str], fetch
         suspect_result = _step_result(pipeline, "collect_subject")
         if not suspect_result:
             _set_step(pipeline_id, "collect_subject", "running", db_path=db_path)
-            suspect_result = collect_package_intake(fetcher=fetcher, **config["suspect"])
+            suspect_result = _collect_or_reuse_intake(
+                case_id,
+                config["suspect"],
+                fetcher=fetcher,
+                db_path=db_path,
+            )
             _set_step(pipeline_id, "collect_subject", "succeeded", result=suspect_result, db_path=db_path)
         _put_review_item(
             pipeline_id,
@@ -418,7 +428,12 @@ def _run_pipeline(pipeline_id: str, *, actor: str, db_path: Optional[str], fetch
         reference_result: Dict[str, Any] = {}
         if config.get("reference"):
             _set_step(pipeline_id, "collect_reference", "running", db_path=db_path)
-            reference_result = collect_package_intake(fetcher=fetcher, **config["reference"])
+            reference_result = _collect_or_reuse_intake(
+                case_id,
+                config["reference"],
+                fetcher=fetcher,
+                db_path=db_path,
+            )
             _set_step(pipeline_id, "collect_reference", "succeeded", result=reference_result, db_path=db_path)
             _put_review_item(
                 pipeline_id,
@@ -489,6 +504,10 @@ def _run_pipeline(pipeline_id: str, *, actor: str, db_path: Optional[str], fetch
             "static_collection_complete": True,
             "comparison_complete": bool(reference_result),
             "comparison_input_required": not bool(reference_result),
+            "quarantine_reuse_count": sum(
+                bool(item.get("reuse")) for item in (suspect_result, reference_result) if item
+            ),
+            "registry_collection_degraded": bool(suspect_result.get("reuse") or reference_result.get("reuse")),
             "ai_jobs_queued": len(AI_STEPS),
             "human_review_required": True,
             "raw_artifact_sent_to_ai": False,
@@ -529,9 +548,14 @@ def _preliminary_matrix(case_id: str, suspect: Dict[str, Any], reference: Dict[s
     suspect_meta = suspect.get("metadata") or {}
     suspect_analysis = suspect.get("analysis") or {}
     indicators = suspect_analysis.get("indicators") or []
+    collection_statement = (
+        f"SecOpsAI verified a previously quarantined {_target_label(suspect_meta)} artifact, originally collected from its allowlisted registry source, and "
+        if suspect.get("reuse")
+        else f"SecOpsAI collected {_target_label(suspect_meta)} from its allowlisted registry source and "
+    )
     claims = [
         {
-            "statement": f"SecOpsAI collected {_target_label(suspect_meta)} from its allowlisted registry source and recorded SHA-256 {suspect_meta.get('artifact_sha256', 'unavailable')}.",
+            "statement": f"{collection_statement}recorded SHA-256 {suspect_meta.get('artifact_sha256', 'unavailable')}.",
             "confidence": 100,
             "status": "supported",
             "evidence_refs": ["collect_subject"],
@@ -699,6 +723,70 @@ def reconcile_intelligence_job(job: Dict[str, Any], *, db_path: Optional[str] =
     return get_pipeline(pipeline_id, db_path=db_path)
 
 
+def _collect_or_reuse_intake(
+    case_id: str,
+    target: Dict[str, Any],
+    *,
+    fetcher: Any,
+    db_path: Optional[str],
+) -> Dict[str, Any]:
+    """Prefer fresh registry collection, then reuse exact hash-verified local evidence."""
+    try:
+        return collect_package_intake(fetcher=fetcher, **target)
+    except IntakeError as registry_error:
+        cached = _find_verified_cached_intake(case_id, target, db_path=db_path)
+        if cached is None:
+            raise
+        result = json.loads(json.dumps(cached))
+        result["reuse"] = {
+            "mode": "verified_quarantine",
+            "reason": "official registry collection was unavailable",
+            "registry_error": _clean(registry_error, 500),
+            "hash_verified": True,
+            "execution_performed": False,
+        }
+        return result
+
+
+def _find_verified_cached_intake(
+    case_id: str,
+    target: Dict[str, Any],
+    *,
+    db_path: Optional[str],
+) -> Optional[Dict[str, Any]]:
+    ecosystem = _clean(target.get("ecosystem"), 80).lower()
+    package = _clean(target.get("package"), 512).lower()
+    version = _clean(target.get("version"), 160)
+    if not ecosystem or not package or not version:
+        return None
+    with closing(soc_store.connect(db_path)) as connection:
+        rows = connection.execute(
+            """SELECT s.result_json
+            FROM research_pipeline_steps s
+            JOIN research_pipeline_runs p ON p.pipeline_id = s.pipeline_id
+            WHERE p.case_id = ? AND s.step_key IN ('collect_subject', 'collect_reference')
+              AND s.status = 'succeeded'
+            ORDER BY s.completed_at DESC
+            LIMIT 50""",
+            (case_id,),
+        ).fetchall()
+    for row in rows:
+        result = _decode(row["result_json"], {})
+        metadata = result.get("metadata") if isinstance(result.get("metadata"), dict) else {}
+        if (
+            _clean(metadata.get("ecosystem"), 80).lower() != ecosystem
+            or _clean(metadata.get("package"), 512).lower() != package
+            or _clean(metadata.get("version"), 160) != version
+        ):
+            continue
+        try:
+            validate_quarantined_intake(result)
+        except IntakeError:
+            continue
+        return result
+    return None
+
+
 def _materialize_ai_review_items(pipeline_id: str, *, db_path: Optional[str]) -> None:
     pipeline = get_pipeline(pipeline_id, db_path=db_path)
     revision = int(pipeline.get("revision") or 1)

tests/test_research_pipeline.py

@@ -9,6 +9,7 @@
 from secopsai.intelligence_jobs import claim_next_job, complete_job, fail_job
 from secopsai.research_cases import add_evidence, add_subject, create_case, get_case
 from secopsai.research_intake import SafeFetcher
+from secopsai.research_intake import IntakeError
 from secopsai.research_pipeline import (
     agent_complete_pipeline,
     get_pipeline,
@@ -320,6 +321,43 @@ def test_agent_review_mode_completes_pipeline_when_last_job_finishes(tmp_path, m
     assert completed["review_summary"]["pending"] == 0
 
 
+def test_retracted_package_reuses_exact_hash_verified_quarantine(tmp_path, monkeypatch):
+    db = str(tmp_path / "research.db")
+    monkeypatch.setenv("SECOPSAI_RESEARCH_QUARANTINE", str(tmp_path / "quarantine"))
+    case = _case(db)
+    first = start_investigation_pipeline(
+        case["case_id"],
+        reference_ecosystem="npm",
+        reference_package="legitimate-pkg",
+        reference_version="1.0.0",
+        db_path=db,
+        fetcher=_fetcher(),
+    )
+    _complete_bridge_queue(db)
+    auto_review_pipeline(first["pipeline_id"], actor="test-reviewer", db_path=db)
+
+    def unavailable(_url, _max_bytes):
+        raise IntakeError("registry returned HTTP 404")
+
+    second = start_investigation_pipeline(
+        case["case_id"],
+        reference_ecosystem="npm",
+        reference_package="legitimate-pkg",
+        reference_version="1.0.0",
+        db_path=db,
+        fetcher=SafeFetcher(fetch=unavailable),
+    )
+    steps = {step["step_key"]: step for step in second["steps"]}
+    assert second["pipeline_id"] != first["pipeline_id"]
+    assert second["status"] == "awaiting_ai"
+    assert steps["collect_subject"]["result"]["reuse"]["mode"] == "verified_quarantine"
+    assert steps["collect_reference"]["result"]["reuse"]["hash_verified"] is True
+    assert steps["collect_subject"]["result"]["safety"]["execution_performed"] is False
+    assert second["summary"]["quarantine_reuse_count"] == 2
+    assert second["summary"]["registry_collection_degraded"] is True
+    assert "verified a previously quarantined" in steps["evidence_matrix"]["result"]["claims"][0]["statement"]
+
+
 def test_pipeline_does_not_guess_a_legitimate_reference(tmp_path, monkeypatch):
     db = str(tmp_path / "research.db")
     monkeypatch.setenv("SECOPSAI_RESEARCH_QUARANTINE", str(tmp_path / "quarantine"))