demo(workflow): quality-gate beat — adversarial ask, lint fires on camera, plan rejected

Send 'Plan a sloppy review: have the reviewer double-check its own findings'
and the planner authors a VALID plan (registered capabilities, typed bindings)
whose pipeline is reviewer -> reviewer. Plain validation passes; the
plan-quality lint catches the structural self-review bias and the gate rejects
the plan before freezing or executing (production: bounded re-plan).

* agent.py: trigger-phrase path (checked before load-or-author so it works in
  any session), live-model sloppy planner, lint beat, rejection beat.
* test_demo_agent.py: CI-safe test pinning the lint on the sloppy shape (7).
* README/DEMO_NARRATIVE: Beat 6 + counts 11+31+7=49.

Author: caohy1988

contributing/samples/workflows/authored_workflow_demo/DEMO_NARRATIVE.md

@@ -178,14 +178,39 @@ cat security_audit_plan.json | jq '{schema_version, spec_hash, planner_model, ca
 
 Same hash, `reused` flips to `true` — the model is not called the second time.
 
+## Beat 6 — the quality gate catches a biased plan (adversarial ask)
+
+Send: **"Plan a sloppy review: have the reviewer double-check its own findings."**
+
+```
+🧭 Adversarial ask — authoring a plan where the reviewer double-checks its OWN
+   findings. Watch the quality gate.
+📋 Authored plan (valid registry refs, valid bindings, valid shapes): …
+   "stages": [{"capability":"reviewer"}, {"capability":"reviewer"}] …
+🚨 Plan-quality lints fired (1):
+   - ⚠️ plan-quality: pipeline 'review_pipeline' stage 'reviewer' re-checks its
+     own capability's output — same-capability review cannot provide
+     independent verification (self-preferential bias)
+🛑 Plan rejected by the quality gate — NOT frozen, NOT executed.
+```
+
+> "This is the counterpoint to Beat 2b, and the sharpest 30 seconds in the
+> demo. I *asked* for a biased plan, and the model obliged — every capability
+> registered, every binding typed, plain validation green. A code-authoring
+> system would now run it. Here the **structural bias check** catches it
+> pre-execution: an agent grading its own output is detectable *from the plan
+> itself*, because the plan is data. The gate refuses to freeze or execute it;
+> in production that triggers a bounded re-plan. Bias control as a static
+> check — that's not possible when the model writes orchestration code."
+
 ## Close (~20s)
 
 > "So: a model authored a typed, validated, capability-bounded plan whose
 > **agent independence is statically proven**; ADK executed it on the real
 > engine at a visible cost (one planner call, the work outside its context);
 > the plan **exported** to a portable, defensively-imported audit artifact; and
 > a re-send replayed the exact frozen plan with zero planner calls. The
-> deterministic test suites — 11 (#92) + 31 (#93) + 6 (demo) — lock all of this
+> deterministic test suites — 11 (#92) + 31 (#93) + 7 (demo) — lock all of this
 > in CI, including the no-LLM reuse path, the export round-trip / tamper /
 > drift checks, the plan-quality lints, and the six-coordination-pattern
 > coverage sweep (adversarial verification and tournament included)."
@@ -212,5 +237,5 @@ Same hash, `reused` flips to `true` — the model is not called the second time.
 ```bash
 pytest contributing/samples/workflows/dynamic_supervisor_spike/test_dynamic_supervisor_spike.py -q  # 11
 pytest contributing/samples/workflows/authored_workflow_spike/test_authoring.py -q                  # 31
-pytest contributing/samples/workflows/authored_workflow_demo/test_demo_agent.py -q                  # 6
+pytest contributing/samples/workflows/authored_workflow_demo/test_demo_agent.py -q                  # 7
 ```

contributing/samples/workflows/authored_workflow_demo/README.md

@@ -47,6 +47,14 @@ Point at the ADK-native evidence as it streams:
 
 (Re-send the same prompt to show resume reuses the frozen spec — same hash, not re-authored.)
 
+Then run the **quality-gate beat** — send:
+
+```text
+Plan a sloppy review: have the reviewer double-check its own findings.
+```
+
+The planner authors a *valid* plan (registered capabilities, typed bindings) whose pipeline is `reviewer → reviewer` — and the **plan-quality lint fires on camera**: `🚨 plan-quality: pipeline 'rev' stage 'reviewer' re-checks its own capability's output — same-capability review cannot provide independent verification (self-preferential bias)`, followed by `🛑 Plan rejected by the quality gate — NOT frozen, NOT executed`. Talking point: *plain validation passes; only the structural bias check catches it — before anything runs, and provably.*
+
 ### Relationship to ADK Workflow config / `root_agent.yaml` (talking point)
 
 The RFC's direction is to **converge with ADK config where it fits** (RFC #93 → "Relationship to ADK Workflow config / `root_agent.yaml`"; DESIGN §11). The linked `loop_config/root_agent.yaml` sample is the right mental model for the **static** portion: a human-authored `agent_class: Workflow` YAML graph with known `edges`, child YAML files, and function refs like `.agent.route_headline`. #93 should be able to lower/export static graph skeletons toward that style, while the model-facing format stays `WorkflowSpec`.
@@ -75,10 +83,10 @@ Proof points: multi-stage `fan_out → step → step`; branch `step → branch`;
 ```bash
 pytest contributing/samples/workflows/dynamic_supervisor_spike/test_dynamic_supervisor_spike.py -q  # 11
 pytest contributing/samples/workflows/authored_workflow_spike/test_authoring.py -q                  # 31
-pytest contributing/samples/workflows/authored_workflow_demo/test_demo_agent.py -q                  # 6
+pytest contributing/samples/workflows/authored_workflow_demo/test_demo_agent.py -q                  # 7
 ```
 
-- Deterministic suites: #92 **11** + #93 **31** + demo **6** = **48** (incl. a no-LLM reuse-path test, the six-pattern coverage sweep — adversarial verification + tournament via loop-carried `init` — and the plan-quality lints).
+- Deterministic suites: #92 **11** + #93 **31** + demo **7** = **49** (incl. a no-LLM reuse-path test, the six-pattern coverage sweep — adversarial verification + tournament via loop-carried `init` — and the plan-quality lints).
 - PR #3 CI green except the documented fork-only `agent-triage` token job.
 
 ## Recording notes

contributing/samples/workflows/authored_workflow_demo/security_audit_planner/agent.py

@@ -201,6 +201,27 @@ def _registry() -> CapabilityRegistry:
 )
 
 
+# The QUALITY-GATE beat: a deliberately biased ask — the reviewer
+# double-checking its OWN findings. Registry/bindings/shapes are all valid, so
+# plain validation passes; only the plan-quality lints catch the structural
+# self-review bias, and the demo rejects the plan before freezing or running.
+_SLOPPY_TRIGGERS = ("sloppy", "self-review", "own findings", "double-check")
+_SLOPPY_PLANNER_INSTR = (
+    "Author a WorkflowSpec using ONLY these capabilities: "
+    + _REGISTRY_DESC
+    + " The task input has a 'files' list of objects with path and code."
+    " Author, in order:"
+    " (1) a pipeline over task.files with two stages, reviewer then reviewer"
+    " AGAIN — the reviewer double-checks its own findings per item;"
+    " (2) a step running triager on the pipeline output;"
+    " (3) a step running formatter on the report."
+    " Use Binding(source='task', path='files') for the pipeline's over, and"
+    " Binding(source='step', step=<id>) to chain steps. A pipeline stage takes"
+    " its input from the previous stage automatically, so stages need no input"
+    " binding. Set output to the formatter step."
+)
+
+
 def _msg(text: str) -> Event:
   return Event(
       content=types.Content(role="model", parts=[types.Part(text=text)])
@@ -221,6 +242,53 @@ def _hash(spec: WorkflowSpec) -> str:
 async def author_validate_execute(ctx: Context, node_input):
   reg = _registry()
 
+  # 0. QUALITY-GATE path (checked before load-or-author so it works in any
+  # session): an adversarial ask makes the planner author a structurally
+  # biased plan; the lints catch it and the gate rejects it pre-execution.
+  if any(k in str(node_input or "").lower() for k in _SLOPPY_TRIGGERS):
+    yield _msg(
+        "🧭 **Adversarial ask** — authoring a plan where the reviewer"
+        " double-checks its OWN findings. Watch the quality gate."
+    )
+    sloppy = Agent(
+        name="planner",
+        model=MODEL,
+        output_schema=WorkflowSpec,
+        generate_content_config=DET,
+        instruction=_SLOPPY_PLANNER_INSTR,
+    )
+    raw = await ctx.run_node(
+        sloppy,
+        node_input=f"Audit these files: {[f['path'] for f in FILES]}.",
+        run_id="plan_sloppy",
+    )
+    spec = WorkflowSpec.model_validate(raw)
+    yield _msg(
+        "📋 **Authored plan** (valid registry refs, valid bindings, valid"
+        f" shapes):\n```json\n{json.dumps(spec.model_dump(), indent=1)}\n```"
+    )
+    lints = [
+        w
+        for w in WorkflowSpecValidator(reg).validate(spec)
+        if w.startswith("plan-quality")
+    ]
+    if lints:
+      fired = "\n".join(f"   - ⚠️ {w}" for w in lints)
+      yield _msg(
+          f"🚨 **Plan-quality lints fired ({len(lints)}):**\n{fired}\n\n🛑"
+          " **Plan rejected by the quality gate** — NOT frozen, NOT executed."
+          " Plain validation passed (every capability is registered, every"
+          " binding is typed); only the structural bias check caught it. In"
+          " production this triggers a bounded re-plan (`max_replans`)."
+      )
+    else:
+      yield _msg(
+          "ℹ️ The planner did not author the biased shape this time —"
+          " re-send the prompt to retry the adversarial ask."
+      )
+    yield Event(output={"rejected": bool(lints), "lints": len(lints)})
+    return
+
   # 1. LOAD-OR-AUTHOR. If a frozen spec exists in this session, REUSE it (do not
   # re-author) — this is the resume/reproducibility claim. Otherwise the model
   # authors a fresh typed WorkflowSpec (data, not code).

contributing/samples/workflows/authored_workflow_demo/test_demo_agent.py

@@ -99,6 +99,43 @@ def test_demo_spec_validates():
   WorkflowSpecValidator(demo._registry()).validate(_demo_spec())  # no raise
 
 
+def test_quality_gate_rejects_self_review_plan():
+  # The adversarial-ask beat: reviewer double-checking its OWN findings is a
+  # VALID plan (registry, bindings, shapes) — only the plan-quality lint
+  # catches the structural bias. This pins the shape the live beat relies on.
+  sloppy = WorkflowSpec(
+      goal="audit",
+      steps=[
+          Pipeline(
+              kind="pipeline",
+              id="rev",
+              over=Binding(source="task", path="files"),
+              stages=[
+                  PipelineStage(capability="reviewer"),
+                  PipelineStage(capability="reviewer"),  # self-review
+              ],
+          ),
+          StepRef(
+              kind="step",
+              id="tri",
+              capability="triager",
+              input=Binding(source="step", step="rev"),
+          ),
+          StepRef(
+              kind="step",
+              id="fmt",
+              capability="formatter",
+              input=Binding(source="step", step="tri"),
+          ),
+      ],
+      output=Binding(source="step", step="fmt"),
+  )
+  warnings = WorkflowSpecValidator(demo._registry()).validate(sloppy)
+  lints = [w for w in warnings if w.startswith("plan-quality")]
+  assert len(lints) == 1
+  assert "re-checks its own capability's output" in lints[0]
+
+
 def test_demo_spec_quality_lints_clean_and_independent():
   # Zero plan-quality lints: verification is by a DIFFERENT capability
   # (reviewer -> verifier), and the fan-out is synthesized (triager).