Replace recursive artifact provenance scanner with explicit declarations (#1160)

* Update jacobian-math skill to remove mode from wire examples

Drop `mode: "EXPLORE"` from the producer wire example, remove `in EXPLORE
mode` / `in VERIFY mode` qualifiers from payload descriptions, and replace
"Use installed VERIFY" with "Use installed checker tools (*.verify,
lean.check, ...)".

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* Replace recursive artifact provenance scanner with explicit declarations (#1108)

AtomicServiceAdapter previously derived CapabilityResult.artifact_uris by
recursively traversing the entire serialized domain output and treating every
string starting with artifact://sha256/ as provenance. A URI in a debug message,
rejected candidate, or example field would be promoted to artifact_uris.

Replace the recursive scanner with an explicit artifact_references callback on
AtomicServiceAdapter. Each adapter registration declares which typed result
fields carry semantically relevant artifact URIs. Verification bindings now
consume explicit references plus record lineage, not string scanning.

- Add artifact_references callback parameter to AtomicServiceAdapter
- Delete _artifact_uris() recursive reflection function
- Migrate all built-in atomic adapter registrations (artifact.put, claim.validate,
  evaluate.batch, witness.find, witness.verify, certificate.verify, shrink.run,
  structure.canonicalize, transform.apply, transform.verify, polytope.separate,
  parameter.region.promote) to declare explicit references
- Add adversarial fixtures proving URIs in debug/rejected/example fields cannot
  enter artifact_uris
- Sync packaged jacobian-math skill with repository copy and update codex
  visibility tests for the prior mode-removal skill edits

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: morluto

npm/skills/jacobian-math/SKILL.md

@@ -22,30 +22,30 @@ Keep decomposition and routing decisions agent-owned; composing already-known
 supporting operations remains allowed when clearer.
 Do not call Jacobian for definitions, formatting, or non-execution tasks.
 
-No discovery for stable producers: `{"capability_id":"<id>","mode":"EXPLORE","payload":<JSON>}`
-(not `COMPUTE`/`input`). Payloads:
+No discovery for stable producers: `{"capability_id":"<id>","input":<JSON>}`.
+Payloads:
 
 - `integer.compute.gcd`, `integer.compute.lcm`, or `integer.compute.extended_gcd`:
   `{"left":"84","right":"30"}`.
 - `matrix.determinant.compute` or `matrix.rank.compute`:
   `{"matrix":{"domain":"QQ","entries":[[{"num":"1","den":"1"}]]}}`.
-- `polynomial.compute.gcd` in `EXPLORE` mode: payload keys are `left` and
+- `polynomial.compute.gcd`: payload keys are `left` and
   `right`; each value has shape
   `{"polynomial_schema_version":"1","domain":"QQ","variables":["x"],"polynomial":{"terms":[{"coefficient":{"num":"1","den":"1"},"exponents":[2]}]}}`.
 - For expression normalization, inspect the known
   `polynomial.expression.normalize` contract directly.
-- `matrix.determinant.verify` in `VERIFY` mode for an independent check:
+- `matrix.determinant.verify` for an independent check:
   `{"determinant_uri":"<determinant_uri from compute output>"}`.
 - `combinatorics.cyclic_difference_set.extension.decide`:
   `{"base_elements":["1","2","4","8","13"],"target_order":7}`.
   `combinatorics.cyclic_difference_set.extension.verify` uses
-  `{"input":<same payload>,"candidate":<producer output.result>}` in `VERIFY` mode.
+  `{"input":<same payload>,"candidate":<producer output.result>}`.
 
 For other outcomes, query `math.find` by plain-language outcome; no ID is
 required. Use low `limit` only for query search, never with `capability_id`. For
 a selected operation's schema, call
-`math.find({"capability_id":"<exact-id>","view":"CONTRACT"})`; never send
-`mode: "CONTRACT"` to `math.run` or put `CONTRACT` in a query. A card's
+`math.find({"capability_id":"<exact-id>","view":"CONTRACT"})`; never put
+`CONTRACT` in a query. A card's
 `invocation_example`, or required top-level fields, may be enough.
 
 Add no domain filter unless its installed spelling is known. Follow exposed
@@ -60,8 +60,9 @@ obligations plus artifact and verification-record URIs.
 Keep representation, decomposition, composition, iteration, verification
 timing, and stopping decisions agent-owned.
 
-Model-authored work is not independent evidence. Use installed `VERIFY` when
-requested; a writable path or schema alone is not authorization. Task-level
+Model-authored work is not independent evidence. Use installed checker tools
+(`*.verify`, `lean.check`, …) when independent verification is requested; a
+writable path or schema alone is not authorization. Task-level
 `VERIFIED` requires exact record bytes, result assurance `VERIFIED`, required
 task authorization and bindings are preserved, and a contract-authorized
 checker identity, digest, or Jacobian record type. Otherwise claim the highest

src/jacobian/atomic_capabilities.py

@@ -79,6 +79,7 @@ def __init__(
         output_schema: dict[str, Any],
         invoke: Callable[[dict[str, Any]], Any],
         store: ArtifactRepository,
+        artifact_references: Callable[[Any], tuple[str, ...]] | None = None,
         unverified_assurance_level: CapabilityAssuranceLevel = (
             CapabilityAssuranceLevel.COMPUTED
         ),
@@ -103,6 +104,7 @@ def __init__(
         )
         self._invoke = invoke
         self._store = store
+        self._artifact_references = artifact_references
         self._unverified_assurance_level = unverified_assurance_level
         self._unverified_basis = unverified_basis
 
@@ -115,7 +117,11 @@ def invoke(self, request: CapabilityRequest) -> CapabilityResult:
         output = _dump(value)
         execution = _execution(value)
         record_uri = _verified_record_uri(value)
-        artifact_uris = _artifact_uris(output)
+        artifact_uris = (
+            self._artifact_references(value)
+            if self._artifact_references is not None
+            else ()
+        )
         verification_artifacts: tuple[str, ...] | None = None
         if record_uri is not None:
             verification_artifacts = _verification_bindings(
@@ -202,6 +208,7 @@ def _adapter(**kwargs: Any) -> AtomicServiceAdapter:
                 parents=tuple(p.get("parents", ())),
                 summary=p.get("summary", ""),
             ),
+            artifact_references=lambda v: (v.artifact_uri,),
             discovery_visible=False,
             tags=("artifact", "storage"),
         ),
@@ -215,6 +222,7 @@ def _adapter(**kwargs: Any) -> AtomicServiceAdapter:
             ),
             output_schema=model_schema(ClaimValidationResult),
             invoke=lambda p: application.claims.validate(**p),
+            artifact_references=lambda v: (v.claim_uri, v.plugin_id),
             read_only=True,
             tags=("claim", "validation"),
         ),
@@ -257,6 +265,11 @@ def _adapter(**kwargs: Any) -> AtomicServiceAdapter:
                     "profile": EvaluationProfile(p["profile"]),
                 }
             ),
+            artifact_references=lambda v: (
+                v.claim_uri,
+                v.plugin_id,
+                *(item.candidate_uri for item in v.items),
+            ),
             unverified_assurance_level=CapabilityAssuranceLevel.HEURISTIC,
             unverified_basis="untrusted plugin evaluation is not independently verified",
             tags=("evaluation",),
@@ -296,6 +309,7 @@ def _adapter(**kwargs: Any) -> AtomicServiceAdapter:
             invoke=lambda p: application.witnesses.find(
                 **{**p, "witness_role": WitnessRole(p["witness_role"])}
             ),
+            artifact_references=lambda v: _witness_find_references(v),
             unverified_assurance_level=CapabilityAssuranceLevel.HEURISTIC,
             unverified_basis="witness search output is evidence pending explicit replay",
             tags=("witness", "search"),
@@ -315,6 +329,7 @@ def _adapter(**kwargs: Any) -> AtomicServiceAdapter:
             ),
             output_schema=model_schema(ResultEnvelope),
             invoke=lambda p: application.verification.verify_witness(**p),
+            artifact_references=lambda v: _envelope_references(v),
             unverified_assurance_level=CapabilityAssuranceLevel.HEURISTIC,
             unverified_basis="the checker did not accept the supplied witness",
             tags=("witness", "verification"),
@@ -329,6 +344,7 @@ def _adapter(**kwargs: Any) -> AtomicServiceAdapter:
             ),
             output_schema=model_schema(ResultEnvelope),
             invoke=lambda p: application.verification.verify_certificate(**p),
+            artifact_references=lambda v: _envelope_references(v),
             unverified_assurance_level=CapabilityAssuranceLevel.HEURISTIC,
             unverified_basis="the checker did not accept the supplied certificate",
             tags=("certificate", "verification"),
@@ -380,6 +396,7 @@ def _adapter(**kwargs: Any) -> AtomicServiceAdapter:
                     "objectives": tuple(p["objectives"]),
                 }
             ),
+            artifact_references=lambda v: _shrink_references(v),
             unverified_assurance_level=CapabilityAssuranceLevel.HEURISTIC,
             unverified_basis="plugin-proposed reductions are not a verified minimality claim",
             tags=("shrink",),
@@ -572,21 +589,36 @@ def _is_verified(value: Any) -> bool:
     return _has_verified_parameter_region_evidence(value)
 
 
-def _artifact_uris(value: Any) -> tuple[str, ...]:
-    found: set[str] = set()
+def _witness_find_references(value: Any) -> tuple[str, ...]:
+    refs: list[str] = [value.claim_uri, value.candidate_uri, value.plugin_id]
+    if value.witness_uri is not None:
+        refs.append(value.witness_uri)
+    if value.certificate_uri is not None:
+        refs.append(value.certificate_uri)
+    return tuple(refs)
 
-    def visit(item: Any) -> None:
-        if isinstance(item, dict):
-            for child in item.values():
-                visit(child)
-        elif isinstance(item, list):
-            for child in item:
-                visit(child)
-        elif isinstance(item, str) and item.startswith("artifact://sha256/"):
-            found.add(item)
 
-    visit(value)
-    return tuple(sorted(found))
+def _envelope_references(value: Any) -> tuple[str, ...]:
+    envelope = _result_envelope(value)
+    if envelope is None:
+        return ()
+    refs: list[str] = list(envelope.evidence_uris)
+    if envelope.verification_record_uri is not None:
+        refs.append(envelope.verification_record_uri)
+    if envelope.assurance.scope_uri is not None:
+        refs.append(envelope.assurance.scope_uri)
+    return tuple(refs)
+
+
+def _shrink_references(value: Any) -> tuple[str, ...]:
+    refs: list[str] = [value.initial_target_uri, value.final_target_uri]
+    for step in value.steps:
+        refs.append(step.from_uri)
+        if step.proposed_uri is not None:
+            refs.append(step.proposed_uri)
+        if step.verification_record_uri is not None:
+            refs.append(step.verification_record_uri)
+    return tuple(refs)
 
 
 def _verification_bindings(

src/jacobian/atomic_domain_capabilities.py

@@ -77,6 +77,7 @@ def build_domain_adapters(
                     ),
                 }
             ),
+            artifact_references=lambda v: _transformation_apply_references(v),
             unverified_assurance_level=CapabilityAssuranceLevel.HEURISTIC,
             unverified_basis=(
                 "plugin transformation output remains an open verification obligation"
@@ -96,6 +97,7 @@ def build_domain_adapters(
             ),
             output_schema=model_schema(ResultEnvelope),
             invoke=lambda p: application.verification.verify_transformation(**p),
+            artifact_references=lambda v: _envelope_references(v),
             unverified_assurance_level=CapabilityAssuranceLevel.HEURISTIC,
             unverified_basis=("the checker did not accept the transformation relation"),
             tags=("transform", "verification"),
@@ -129,6 +131,7 @@ def build_domain_adapters(
             invoke=lambda p: application.polytope.separate(
                 PolytopeSeparateRequest(**p)
             ),
+            artifact_references=lambda v: _polytope_separate_references(v),
             tags=("polytope", "exact"),
             provider="jacobian.z3",
         ),
@@ -148,7 +151,55 @@ def build_domain_adapters(
             ),
             output_schema=model_schema(ParameterRegion),
             invoke=lambda p: application.conjectures.promote_parameter_region(**p),
+            artifact_references=lambda v: _parameter_region_references(v),
             read_only=True,
             tags=("parameter", "verification"),
         ),
     )
+
+
+def _transformation_apply_references(value: Any) -> tuple[str, ...]:
+    refs: list[str] = [value.source_uri]
+    if value.target_uri is not None:
+        refs.append(value.target_uri)
+    if value.claim_uri is not None:
+        refs.append(value.claim_uri)
+    if value.transformation_uri is not None:
+        refs.append(value.transformation_uri)
+    return tuple(refs)
+
+
+def _envelope_references(value: Any) -> tuple[str, ...]:
+    envelope = getattr(value, "result", None)
+    if not isinstance(envelope, ResultEnvelope):
+        return ()
+    refs: list[str] = list(envelope.evidence_uris)
+    if envelope.verification_record_uri is not None:
+        refs.append(envelope.verification_record_uri)
+    if envelope.assurance.scope_uri is not None:
+        refs.append(envelope.assurance.scope_uri)
+    return tuple(refs)
+
+
+def _polytope_separate_references(value: Any) -> tuple[str, ...]:
+    refs: list[str] = [value.point_uri, value.generator_set_uri]
+    if value.effective_point_uri is not None:
+        refs.append(value.effective_point_uri)
+    if value.effective_generator_set_uri is not None:
+        refs.append(value.effective_generator_set_uri)
+    if value.claim_uri is not None:
+        refs.append(value.claim_uri)
+    if value.witness_uri is not None:
+        refs.append(value.witness_uri)
+    if value.certificate_uri is not None:
+        refs.append(value.certificate_uri)
+    return tuple(refs)
+
+
+def _parameter_region_references(value: Any) -> tuple[str, ...]:
+    refs: list[str] = list(value.sample_uris)
+    if value.subject_uri is not None:
+        refs.append(value.subject_uri)
+    if value.verification_record_uri is not None:
+        refs.append(value.verification_record_uri)
+    return tuple(refs)

src/jacobian/atomic_experiment_capabilities.py

@@ -61,6 +61,7 @@ def build_experiment_adapters(
             ),
             output_schema=model_schema(StructureCanonicalizationResult),
             invoke=lambda p: application.structures.canonicalize(**p),
+            artifact_references=lambda v: _structure_references(v),
             unverified_assurance_level=CapabilityAssuranceLevel.HEURISTIC,
             unverified_basis="plugin canonicalization is not independently verified",
             tags=("structure", "canonicalization"),
@@ -154,3 +155,10 @@ def build_experiment_adapters(
             tags=("experiment", "control"),
         ),
     )
+
+
+def _structure_references(value: Any) -> tuple[str, ...]:
+    refs: list[str] = [value.structure_uri]
+    if value.canonical_uri is not None:
+        refs.append(value.canonical_uri)
+    return tuple(refs)

tests/component/capabilities/test_atomic_capabilities.py

@@ -1,5 +1,7 @@
 from pathlib import Path
 
+from pydantic import BaseModel
+
 from jacobian.atomic_capabilities import AtomicServiceAdapter
 from jacobian.contracts.capabilities import (
     CapabilityCompletenessStatus,
@@ -20,6 +22,8 @@
 )
 from jacobian.storage.repository import ArtifactRepository
 
+_ARTIFACT_URI = "artifact://sha256/" + "a" * 64
+
 
 def test_failed_exhaustive_result_cannot_claim_complete_coverage(
     tmp_path: Path,
@@ -96,3 +100,79 @@ def test_exhaustive_result_without_scope_cannot_claim_complete_coverage(
     assert result.execution.status is ExecutionStatus.COMPLETED
     assert result.scope is None
     assert result.completeness.status is CapabilityCompletenessStatus.PARTIAL
+
+
+def test_artifact_uri_in_explanatory_field_is_not_promoted_to_provenance(
+    tmp_path: Path,
+) -> None:
+    """A valid artifact URI in a debug/message field must not enter artifact_uris."""
+
+    class FakeResult(BaseModel):
+        artifact_uri: str = _ARTIFACT_URI
+        debug_message: str = f"{_ARTIFACT_URI} was considered but not used"
+
+    adapter = AtomicServiceAdapter(
+        capability_id="test.explicit",
+        title="Test explicit references",
+        description="Only the declared artifact_uri should appear.",
+        input_schema={"type": "object", "additionalProperties": False},
+        output_schema={"type": "object"},
+        invoke=lambda _payload: FakeResult(),
+        store=ArtifactRepository(tmp_path),
+        artifact_references=lambda v: (v.artifact_uri,),
+    )
+
+    result = adapter.invoke(CapabilityRequest(capability_id="test.explicit", input={}))
+
+    assert result.artifact_uris == (_ARTIFACT_URI,)
+
+
+def test_no_artifact_references_callback_yields_empty_artifact_uris(
+    tmp_path: Path,
+) -> None:
+    """Without an artifact_references callback, artifact_uris must be empty."""
+
+    class FakeResult(BaseModel):
+        explanatory: str = f"see {_ARTIFACT_URI} for context"
+
+    adapter = AtomicServiceAdapter(
+        capability_id="test.no_refs",
+        title="Test no references",
+        description="No artifact references declared.",
+        input_schema={"type": "object", "additionalProperties": False},
+        output_schema={"type": "object"},
+        invoke=lambda _payload: FakeResult(),
+        store=ArtifactRepository(tmp_path),
+    )
+
+    result = adapter.invoke(CapabilityRequest(capability_id="test.no_refs", input={}))
+
+    assert result.artifact_uris == ()
+
+
+def test_rejected_candidate_uri_does_not_enter_artifact_uris(
+    tmp_path: Path,
+) -> None:
+    """A rejected candidate URI must not become provenance."""
+
+    class FakeResult(BaseModel):
+        produced_uri: str = _ARTIFACT_URI
+        rejected_candidates: tuple[str, ...] = (
+            "artifact://sha256/" + "b" * 64,
+            "artifact://sha256/" + "c" * 64,
+        )
+
+    adapter = AtomicServiceAdapter(
+        capability_id="test.rejected",
+        title="Test rejected candidates",
+        description="Only produced_uri should appear.",
+        input_schema={"type": "object", "additionalProperties": False},
+        output_schema={"type": "object"},
+        invoke=lambda _payload: FakeResult(),
+        store=ArtifactRepository(tmp_path),
+        artifact_references=lambda v: (v.produced_uri,),
+    )
+
+    result = adapter.invoke(CapabilityRequest(capability_id="test.rejected", input={}))
+
+    assert result.artifact_uris == (_ARTIFACT_URI,)