feat(ca-demo): frozen middle results record their workflow lineage

User question exposed the gap: middle results were stored BESIDE the
frozen workflow (linked only by the inspector's heuristic), not
structurally attached to it. The artifact now records plan_hash and
produced_by_step at freeze time; revisions inherit the lineage. Design
kept deliberate: the plan envelope stays a question-agnostic TEMPLATE
(embedding per-question instances would bloat and conflate it) — middle
results are per-task-input instances that REFERENCE their plan. The
inspector attaches artifacts to workflow cards by recorded lineage
(heuristic fallback for pre-lineage records) and renders the lineage on
each card. Lineage pinned by tests incl. revision inheritance.

Author: caohy1988

contributing/samples/workflows/authored_workflow_ca_demo/bq_ca_planner/agent.py

@@ -1444,6 +1444,8 @@ def _freeze_sql(
     bytes_processed: int = 0,
     feedback: str | None = None,
     previous: dict | None = None,
+    plan_hash: str | None = None,
+    produced_by_step: str | None = None,
 ) -> str:
   """Freeze (or revise) the validated SQL for a question.
 
@@ -1476,6 +1478,14 @@ def _freeze_sql(
       "bytes_processed": int(bytes_processed or 0),
       "validated_at": _now_iso(),
   })
+  # WORKFLOW LINEAGE: the middle result is an instance of a specific plan's
+  # step for a specific question — record which plan (hash) and which step
+  # produced it, so the artifact is structurally attached to the frozen
+  # workflow, not just stored beside it. Revisions inherit the lineage.
+  if plan_hash:
+    rec["plan_hash"] = plan_hash
+  if produced_by_step:
+    rec["produced_by_step"] = produced_by_step
   with open(path, "w") as f:
     json.dump(rec, f, indent=1)
   return path
@@ -1836,20 +1846,22 @@ async def plan_and_run(ctx: Context, node_input):
     # a frozen plan — freeze its dry-run-validated output so replays of
     # this exact question are numerically deterministic (and feedback can
     # amend it auditably).
-    checked = next(
+    checked_step, checked = next(
         (
-            v
-            for v in interp.state.values()
+            (k, v)
+            for k, v in interp.state.items()
             if isinstance(v, dict) and v.get("valid") is True and v.get("sql")
         ),
-        None,
+        (None, None),
     )
     if checked:
       _freeze_sql(
           task["question"],
           checked["sql"],
           engine=str(checked.get("engine", "bigquery")),
           bytes_processed=checked.get("bytes_processed", 0),
+          plan_hash=spec_hash,
+          produced_by_step=checked_step,
       )
       yield _msg(
           "🧊 **SQL frozen** for this question — re-ask it (any session)"

contributing/samples/workflows/authored_workflow_ca_demo/plan_inspector.py

@@ -176,17 +176,31 @@ def _kv(label, value, why, tag, tag_label) -> str:
   )
 
 
-def _sql_producing_step_ids(spec) -> list:
-  """The plan steps whose validated results the store freezes (the drafting
-  loop in the ask-a-question plan)."""
-  ids = []
+def _sql_producing_step_ids(spec, mid_results=()) -> list:
+  """The plan steps whose validated results froze. Prefer the lineage the
+  artifacts RECORDED (produced_by_step); fall back to the drafting-loop
+  heuristic for records frozen before lineage existed."""
+  recorded = {
+      r.get("produced_by_step")
+      for r in mid_results
+      if r.get("produced_by_step")
+  }
+  step_ids = {s.get("id") for s in spec.get("steps", [])}
+  # lineage may point INSIDE a loop body — surface the loop in that case
   for s in spec.get("steps", []):
-    sid = str(s.get("id", "")).lower()
-    if s.get("kind") == "loop_until" and any(
-        h in sid for h in _SQL_PRODUCER_HINTS
-    ):
-      ids.append(s.get("id"))
-  return ids
+    if s.get("kind") == "loop_until":
+      body_ids = {b.get("id") for b in s.get("body", [])}
+      if recorded & body_ids:
+        recorded.add(s.get("id"))
+  ids = [i for i in recorded if i in step_ids]
+  if ids:
+    return ids
+  return [
+      s.get("id")
+      for s in spec.get("steps", [])
+      if s.get("kind") == "loop_until"
+      and any(h in str(s.get("id", "")).lower() for h in _SQL_PRODUCER_HINTS)
+  ]
 
 
 def _mid_result(rec: dict) -> str:
@@ -206,8 +220,16 @@ def _mid_result(rec: dict) -> str:
       f" <code>{rec.get('sql_hash', '')[:16]}</code> · validated"
       f" {str(rec.get('validated_at', ''))[:19]} ·"
       f" engine {html.escape(str(rec.get('engine', '')))} ·"
-      f" {len(revs)} human revision(s)</div>"
-      f"<details><summary>the validated artifact (SQL, in this instance)"
+      f" {len(revs)} human revision(s)"
+      + (
+          " · <b>lineage:</b> plan"
+          f" <code>{html.escape(str(rec['plan_hash']))}</code>, step"
+          f" <code>{html.escape(str(rec.get('produced_by_step', '?')))}</code>"
+          if rec.get("plan_hash")
+          else ""
+      )
+      + "</div>"
+      "<details><summary>the validated artifact (SQL, in this instance)"
       f"</summary><pre>{html.escape(rec.get('sql', ''))}</pre></details>"
       + (rev_html or "")
       + "</div>"
@@ -216,7 +238,7 @@ def _mid_result(rec: dict) -> str:
 
 def _plan_card(name: str, env: dict, mid_results=()) -> str:
   spec = env.get("spec", {})
-  frozen_ids = _sql_producing_step_ids(spec) if mid_results else []
+  frozen_ids = _sql_producing_step_ids(spec, mid_results) if mid_results else []
   parts = [
       f'<div class="card"><h2 style="margin-top:0">Frozen workflow:'
       f" {html.escape(name)} — “{html.escape(spec.get('goal', ''))}”</h2>",
@@ -440,7 +462,13 @@ def main() -> str:
 
   cards = timeline
   for name, env in envs.items():
-    attach = in_session if name == "sequence" else ()
+    short = str(env.get("spec_hash", ""))[:12]
+    attach = [
+        r
+        for r in in_session
+        if r.get("plan_hash") == short
+        or (not r.get("plan_hash") and name == "sequence")
+    ]
     cards += _plan_card(name, env, attach)
   if other:
     cards += (

contributing/samples/workflows/authored_workflow_ca_demo/test_ca_demo_agent.py

@@ -1139,6 +1139,22 @@ def test_sql_freezing_roundtrip_and_revision_history(tmp_path, monkeypatch):
   assert "calendar quarters" in rec2["revisions"][0]["feedback"]
   # unknown question -> None
   assert demo._load_frozen_sql("never asked") is None
+  # WORKFLOW LINEAGE: the middle result records which plan + step produced
+  # it, and revisions inherit the lineage (the artifact is structurally
+  # attached to the frozen workflow, not just stored beside it).
+  demo._freeze_sql(
+      "lineage q",
+      "SELECT 1",
+      plan_hash="abc123def456",
+      produced_by_step="sqlgen",
+  )
+  rec3 = demo._load_frozen_sql("lineage q")
+  assert rec3["plan_hash"] == "abc123def456"
+  assert rec3["produced_by_step"] == "sqlgen"
+  demo._freeze_sql("lineage q", "SELECT 2", feedback="tweak it", previous=rec3)
+  rec4 = demo._load_frozen_sql("lineage q")
+  assert rec4["plan_hash"] == "abc123def456"  # lineage survives revisions
+  assert len(rec4["revisions"]) == 1
 
 
 def test_frozen_sql_replay_plan_is_static_and_clean():