fix(history): count cached re-runs of runtime-linked text as tracked runs (#403)

Since #347 a cached re-run of runtime-linked prompt text (PreviewAny or an
LLM into a PromptManager text input) counts in the Library, but its history
capture had no raw text, so the history page's "tracked runs" stayed at the
number of jobs in which the node executed.

The #347 receipt now also keeps the raw, prepend and append text the node
gave the history capture. When a cached node's graph inference leaves a field
unknown, the capture takes it from the receipt that matches the cache entry,
labeled inferred/graph_semantics, so the job joins the prompt's history scope
and gets its prompt hint. No schema change and no new evidence vocabulary;
no backfill. Fails closed on no receipt, a disagreeing known field, equal
outputs with different fields, oversized or invalid text and a full budget
(least recently used receipts give up their fields first; counting stays).

utils/execution_context_host.py

@@ -151,10 +151,77 @@ def cached_string_outputs(
         return None
 
 
+def _receipt_fields(
+    receipts: object, executor: object, job_id: str, source: object
+) -> list | None:
+    """Per list index, the ``(raw, prepend, append)`` of the execution reused (#403).
+
+    From the #347 receipt whose digests equal every STRING slot of the cache
+    entry; None when there is none or the answer is malformed.
+    """
+    try:
+        if receipts is None:
+            try:
+                from .cached_runs import CACHED_RUNS as receipts
+            except ImportError:
+                from utils.cached_runs import CACHED_RUNS as receipts
+        values = cached_string_outputs(executor, source.node_id, source.node_type)
+        if values is None:
+            return None
+        found = receipts.recorded_fields(job_id, source.node_id, values)
+        if type(found) is not list or len(found) != len(values[0]):
+            return None
+        position = (
+            (1 if source.output_slot in (1, 3) else 0)
+            if source.node_type == "PromptManagerPair"
+            else 0
+        )
+        fields = []
+        for slots in found:
+            slot = slots[position] if type(slots) is tuple else None
+            if not (
+                type(slot) is tuple
+                and len(slot) == 3
+                and all(type(value) is str for value in slot)
+            ):
+                return None
+            fields.append(slot)
+        return fields
+    except Exception:
+        return None
+
+
+def _agrees(inferred: PromptSnapshot, fields: tuple) -> bool:
+    """Every field the graph knows equals the receipt's; else fill nothing."""
+    known = (inferred.raw, inferred.prepend, inferred.append)
+    return all(
+        field.value is None or field.value == value
+        for field, value in zip(known, fields)
+    )
+
+
+def _filled(inferred: TextEvidence, value: str) -> TextEvidence:
+    """A field the graph cannot show, taken from the reused execution (#403)."""
+    if inferred.value is not None:
+        return inferred
+    return TextEvidence(value, "inferred", "graph_semantics", None)
+
+
 def observe_cached_context(
-    executor: object, message_type: object, message_data: object
+    executor: object,
+    message_type: object,
+    message_data: object,
+    *,
+    receipts: object = None,
 ) -> None:
-    """Copy bounded local strings without cache/provider execution or retention."""
+    """Copy bounded local strings without cache/provider execution or retention.
+
+    Raw, prepend and append come from graph inference; a field it leaves
+    unknown (runtime-only text) is taken from the receipt of the execution the
+    cache entry is proven to come from (``CachedRuns.recorded_fields``), so the
+    cached job is a tracked run of its prompt (#403). ``receipts`` defaults to
+    the process ``CACHED_RUNS``.
+    """
     try:
         binding = active_capture()
         if binding is None or type(message_type) is not str:
@@ -198,18 +265,29 @@ def observe_cached_context(
             inferred = manager.source_snapshot(cap, source)
             if inferred is None:
                 continue
+            fields = None
+            if any(
+                field.value is None
+                for field in (inferred.raw, inferred.prepend, inferred.append)
+            ):
+                fields = _receipt_fields(receipts, executor, cap.job_id, source)
             for index, value in enumerate(values):
                 if type(value) is not str:
                     continue
+                raw, prepend, append = inferred.raw, inferred.prepend, inferred.append
+                if fields is not None and _agrees(inferred, fields[index]):
+                    raw = _filled(raw, fields[index][0])
+                    prepend = _filled(prepend, fields[index][1])
+                    append = _filled(append, fields[index][2])
                 manager.observe(
                     cap,
                     source,
                     index,
                     PromptSnapshot(
-                        inferred.raw,
+                        raw,
                         TextEvidence(value, "recorded", "host_cache_output", None),
-                        inferred.prepend,
-                        inferred.append,
+                        prepend,
+                        append,
                     ),
                 )
     except Exception: