diff --git a/fern/openapi/skyvern_openapi.json b/fern/openapi/skyvern_openapi.json
index 9b9ed91cc..7dd179cbb 100644
--- a/fern/openapi/skyvern_openapi.json
+++ b/fern/openapi/skyvern_openapi.json
@@ -15634,6 +15634,11 @@
             ],
             "title": "Workflow Permanent Id"
           },
+          "workflow_deleted": {
+            "type": "boolean",
+            "title": "Workflow Deleted",
+            "default": false
+          },
           "script_run": {
             "type": "boolean",
             "title": "Script Run",
diff --git a/skyvern/client/types/task_run_list_item.py b/skyvern/client/types/task_run_list_item.py
index 19b3009b5..f5b629da2 100644
--- a/skyvern/client/types/task_run_list_item.py
+++ b/skyvern/client/types/task_run_list_item.py
@@ -21,6 +21,7 @@ class TaskRunListItem(UniversalBaseModel):
     finished_at: typing.Optional[dt.datetime] = None
     created_at: dt.datetime
     workflow_permanent_id: typing.Optional[str] = None
+    workflow_deleted: bool = False
     script_run: typing.Optional[bool] = None
 
     if IS_PYDANTIC_V2:
diff --git a/skyvern/forge/prompts/skyvern/script-reviewer-extraction.j2 b/skyvern/forge/prompts/skyvern/script-reviewer-extraction.j2
index b51a0394d..baaaa929d 100644
--- a/skyvern/forge/prompts/skyvern/script-reviewer-extraction.j2
+++ b/skyvern/forge/prompts/skyvern/script-reviewer-extraction.j2
@@ -59,6 +59,11 @@ These are the KNOWN parameter names for `context.parameters[...]`:
 {% endfor %}
 For fields not covered by these parameters, use `ai='proactive'` with a descriptive prompt (see Rule 8b).
 {% endif %}
+{% if self_output_key %}
+
+### Forbidden parameter key for this block
+- `{{ self_output_key }}` — this block's own output slot. It is `None` until the block has finished running, so `context.parameters['{{ self_output_key }}']` crashes inside this block with `'NoneType' object is not subscriptable`. Upstream blocks' `<label>_output` keys ARE valid; only this block's own output is forbidden.
+{% endif %}
 {% if run_parameter_values %}
 
 ## Current Run Parameter Values
diff --git a/skyvern/forge/prompts/skyvern/script-reviewer-form-filling.j2 b/skyvern/forge/prompts/skyvern/script-reviewer-form-filling.j2
index c28b30a8f..5acf33b46 100644
--- a/skyvern/forge/prompts/skyvern/script-reviewer-form-filling.j2
+++ b/skyvern/forge/prompts/skyvern/script-reviewer-form-filling.j2
@@ -65,6 +65,11 @@ These are the KNOWN parameter names for `context.parameters[...]`:
 {% endfor %}
 For fields not covered by these parameters, use `"ai": "proactive"` with a descriptive prompt (see Rule 9b).
 {% endif %}
+{% if self_output_key %}
+
+### Forbidden parameter key for this block
+- `{{ self_output_key }}` — this block's own output slot. It is `None` until the block has finished running, so `context.parameters['{{ self_output_key }}']` crashes inside this block with `'NoneType' object is not subscriptable`. Upstream blocks' `<label>_output` keys ARE valid; only this block's own output is forbidden.
+{% endif %}
 {% if run_parameter_values %}
 
 ## Current Run Parameter Values
diff --git a/skyvern/forge/prompts/skyvern/script-reviewer.j2 b/skyvern/forge/prompts/skyvern/script-reviewer.j2
index 69a0e05c6..ef806aa7b 100644
--- a/skyvern/forge/prompts/skyvern/script-reviewer.j2
+++ b/skyvern/forge/prompts/skyvern/script-reviewer.j2
@@ -94,6 +94,11 @@ These are the KNOWN parameter names for `context.parameters[...]`:
 {% endfor %}
 For fields not covered by these parameters: use Python code if the value is deterministic (see "Deterministic Logic"), or `ai='proactive'` if it requires subjective judgment (see Rule 9c).
 {% endif %}
+{% if self_output_key %}
+
+### Forbidden parameter key for this block
+- `{{ self_output_key }}` — this block's own output slot. It is `None` until the block has finished running, so `context.parameters['{{ self_output_key }}']` (or any subscript such as `["username"]`, `["password"]`) crashes inside this block with `'NoneType' object is not subscriptable`. Upstream blocks' `<label>_output` keys ARE valid; only this block's own output is forbidden. If you need the user's credentials or a similar value, use a credential parameter from the list above or `ai='proactive'`.
+{% endif %}
 {% if run_parameter_values %}
 
 ## Current Run Parameter Values
diff --git a/skyvern/forge/sdk/db/repositories/workflow_runs.py b/skyvern/forge/sdk/db/repositories/workflow_runs.py
index 59d6b50a5..ed3cb2615 100644
--- a/skyvern/forge/sdk/db/repositories/workflow_runs.py
+++ b/skyvern/forge/sdk/db/repositories/workflow_runs.py
@@ -516,6 +516,17 @@ class WorkflowRunsRepository(BaseRepository):
     ) -> list[dict[str, Any]]:
         async with self.Session() as session:
             effective_status = func.coalesce(WorkflowRunModel.status, TaskRunModel.status)
+            # True iff this row's workflow_permanent_id has no active (deleted_at IS NULL) version.
+            workflow_deleted_expr = and_(
+                TaskRunModel.workflow_permanent_id.isnot(None),
+                ~exists().where(
+                    and_(
+                        WorkflowModel.workflow_permanent_id == TaskRunModel.workflow_permanent_id,
+                        WorkflowModel.organization_id == TaskRunModel.organization_id,
+                        WorkflowModel.deleted_at.is_(None),
+                    )
+                ),
+            ).label("workflow_deleted")
             query = (
                 select(
                     TaskRunModel.task_run_id.label("task_run_id"),
@@ -529,6 +540,7 @@ class WorkflowRunsRepository(BaseRepository):
                     TaskRunModel.workflow_permanent_id.label("workflow_permanent_id"),
                     TaskRunModel.script_run.label("script_run"),
                     TaskRunModel.searchable_text.label("searchable_text"),
+                    workflow_deleted_expr,
                 )
                 .select_from(TaskRunModel)
                 .outerjoin(
diff --git a/skyvern/schemas/runs.py b/skyvern/schemas/runs.py
index b89f36724..2ffb62b84 100644
--- a/skyvern/schemas/runs.py
+++ b/skyvern/schemas/runs.py
@@ -740,6 +740,7 @@ class TaskRunListItem(BaseModel):
     finished_at: datetime | None = None
     created_at: datetime
     workflow_permanent_id: str | None = None
+    workflow_deleted: bool = False
     script_run: bool = False
     searchable_text: str | None = Field(default=None, exclude=True)
 
diff --git a/skyvern/services/script_reviewer.py b/skyvern/services/script_reviewer.py
index c0faf1166..60b6f32da 100644
--- a/skyvern/services/script_reviewer.py
+++ b/skyvern/services/script_reviewer.py
@@ -384,6 +384,14 @@ class ScriptReviewer:
         if not conditional_episodes:
             return None
 
+        # Conditional code is persisted via `create_script_version_from_review`
+        # too, so it must run the same parameter-reference validation as the
+        # regular reviewer path. Load the workflow's parameter keys here.
+        _, all_parameter_keys, _ = await self._load_workflow_context(
+            organization_id=organization_id,
+            workflow_permanent_id=workflow_permanent_id,
+        )
+
         # Group episodes by block label — keep only the latest per block
         latest_by_block: dict[str, ScriptFallbackEpisode] = {}
         for ep in conditional_episodes:
@@ -399,6 +407,7 @@ class ScriptReviewer:
                     episode=episode,
                     organization_id=organization_id,
                     run_parameter_values=run_parameter_values,
+                    all_parameter_keys=all_parameter_keys,
                 )
                 if code:
                     updated_blocks[block_label] = code
@@ -415,6 +424,7 @@ class ScriptReviewer:
         block_label: str,
         episode: ScriptFallbackEpisode,
         organization_id: str,
+        all_parameter_keys: list[str],
         run_parameter_values: dict[str, str] | None = None,
     ) -> str | None:
         """Generate Python code for a conditional block based on its expression patterns.
@@ -560,6 +570,29 @@ class ScriptReviewer:
                         current_prompt = self._build_retry_prompt(code, hardcoded_error, function_signature)
                     continue
 
+                # The filter strips `<block_label>_output` from the valid set so
+                # the validator's primary mode rejects any ref to the current
+                # block's own output. `self_output_key` is also threaded through
+                # to (a) catch the empty-valid-set edge case and (b) augment the
+                # retry hint with the NoneType crash explanation.
+                self_output_key = f"{block_label}_output" if block_label else None
+                conditional_param_keys = self._filter_self_output_param_keys(
+                    sorted(set(all_parameter_keys)), block_label or ""
+                )
+                param_error = self._validate_parameter_references(
+                    code, conditional_param_keys, self_output_key=self_output_key
+                )
+                if param_error is not None:
+                    LOG.warning(
+                        "ScriptReviewer: conditional code has invalid parameter references",
+                        block_label=block_label,
+                        attempt=attempt,
+                        error=param_error,
+                    )
+                    if attempt < max_attempts:
+                        current_prompt = self._build_retry_prompt(code, param_error, function_signature)
+                    continue
+
                 LOG.info(
                     "ScriptReviewer: generated conditional code",
                     block_label=block_label,
@@ -731,6 +764,14 @@ class ScriptReviewer:
         # and merge with workflow-level parameter keys for a complete list.
         goal_param_keys = set(re.findall(r"\{\{\s*(\w+)\s*\}\}", navigation_goal))
         parameter_keys = sorted(goal_param_keys | set(all_parameter_keys or []))
+        # The filter strips `<block_label>_output` from the valid set so the
+        # validator's primary mode rejects any ref to the current block's own
+        # output (it's None until the block finishes — `'NoneType' object is
+        # not subscriptable'`). `self_output_key` is also threaded into the
+        # validator and prompt template to (a) catch the empty-valid-set edge
+        # case and (b) explain the crash mode in retry hints.
+        self_output_key = f"{block_label}_output" if block_label else None
+        parameter_keys = self._filter_self_output_param_keys(parameter_keys, block_label or "")
 
         # Build historical episode summaries for cross-run context.
         # Include per-run parameter values so the reviewer can detect that
@@ -782,6 +823,7 @@ class ScriptReviewer:
             terminate_criterion=terminate_criterion,
             complete_criterion=complete_criterion,
             error_code_mapping=error_code_mapping,
+            self_output_key=self_output_key,
         )
 
         LOG.info(
@@ -918,8 +960,11 @@ class ScriptReviewer:
                             current_prompt = self._build_retry_prompt(updated_code, classify_error, function_signature)
                         continue
 
-                # Validate parameter references (catch invented context.parameters['...'] keys)
-                param_error = self._validate_parameter_references(updated_code, parameter_keys)
+                # Validate parameter references — invented keys and the
+                # current block's own `_output` ref both fail here.
+                param_error = self._validate_parameter_references(
+                    updated_code, parameter_keys, self_output_key=self_output_key
+                )
                 if param_error is not None:
                     LOG.warning(
                         "ScriptReviewer: invalid parameter reference, retrying",
@@ -1656,50 +1701,118 @@ class ScriptReviewer:
             return None
         return "; ".join(errors[:3])
 
-    # Regex to find context.parameters['key'] or context.parameters["key"]
-    _PARAM_REF_RE = re.compile(r"""context\.parameters\[['"](\w+)['"]\]""")
+    # Matches both subscript (`context.parameters['key']` / `[ 'key' ]`) and
+    # dict-get (`context.parameters.get('key')` / `.get('key', default)`) access.
+    # `\s*` at every accessor boundary so reformatted variants (newlines inside
+    # the subscript, spaces between `parameters` and `.get`, etc.) can't bypass
+    # the validator. A libcst AST walker would be a stronger replacement —
+    # `_validate_no_hardcoded_values` already uses libcst as precedent.
+    _PARAM_REF_RE = re.compile(
+        r"""context\s*\.\s*parameters\s*(?:\[\s*['"](\w+)['"]\s*\]|\.\s*get\s*\(\s*['"](\w+)['"]\s*(?:,[^)]*)?\))"""
+    )
 
-    def _find_param_refs_excluding_comments(self, code: str) -> list[str]:
-        """Extract parameter reference keys from code, skipping comment lines."""
-        refs: list[str] = []
+    @staticmethod
+    def _strip_comment_lines(code: str) -> str:
+        out = []
         for line in code.split("\n"):
             if line.lstrip().startswith("#"):
-                continue
-            for match in self._PARAM_REF_RE.finditer(line):
-                refs.append(match.group(1))
+                out.append("")
+            else:
+                out.append(line)
+        return "\n".join(out)
+
+    def _find_param_refs_excluding_comments(self, code: str) -> list[str]:
+        """Extract parameter reference keys from code, skipping comment lines.
+
+        Uses a whole-code regex scan (rather than per-line) so multiline
+        access like::
+
+            value=context.parameters[
+                'block_output'
+            ]['username']
+
+        is still matched. The regex's ``\\s*`` already permits newlines inside
+        the subscript brackets and `.get()` parens.
+        """
+        scrubbed = self._strip_comment_lines(code)
+        refs: list[str] = []
+        for match in self._PARAM_REF_RE.finditer(scrubbed):
+            # Group 1 = subscript form, group 2 = .get() form; exactly one fires.
+            key = match.group(1) or match.group(2)
+            if key:
+                refs.append(key)
         return refs
 
-    def _validate_parameter_references(self, code: str, parameter_keys: list[str]) -> str | None:
+    @staticmethod
+    def _filter_self_output_param_keys(parameter_keys: list[str], block_label: str) -> list[str]:
+        """Drop the current block's own ``<block_label>_output`` from the parameter list.
+
+        The output slot is ``None`` until the block has finished running, so any
+        ``context.parameters['<self>_output']`` reference inside the block crashes
+        with ``'NoneType' object is not subscriptable``. Upstream and downstream
+        block outputs stay — only the current block's output is forbidden.
+        """
+        if not block_label:
+            return parameter_keys
+        self_output_key = f"{block_label}_output"
+        return [k for k in parameter_keys if k != self_output_key]
+
+    def _validate_parameter_references(
+        self,
+        code: str,
+        parameter_keys: list[str],
+        self_output_key: str | None = None,
+    ) -> str | None:
         """Validate that context.parameters['key'] references use known parameter keys.
 
         Catches KeyError crashes at runtime when the LLM invents parameter names.
-        Returns an error message or None if all references are valid.
+        When ``self_output_key`` is supplied and matches one of the rejected
+        references, the retry hint explains the self-output ``None``-subscript
+        crash so the LLM doesn't reintroduce the same pattern.
+
+        Short-circuits on an empty valid set only when ``self_output_key`` is
+        also absent. With a self-output key in play we still scan so the bug
+        can't slip through workflows whose only declared parameter was the
+        block's own output.
         """
-        if not parameter_keys:
-            return None  # No parameter keys to validate against
+        if not parameter_keys and not self_output_key:
+            return None  # No constraints to validate against (preserves prior behavior).
 
         valid_keys = set(parameter_keys)
         invalid: list[str] = []
 
-        for line in code.split("\n"):
-            stripped = line.lstrip()
-            if stripped.startswith("#"):
+        # Empty valid set + ``self_output_key`` rejects only the self-output
+        # ref; other unknown refs may be runtime-valid synthesized keys we
+        # don't load into the valid set, so we let them through.
+        scrubbed = self._strip_comment_lines(code)
+        for match in self._PARAM_REF_RE.finditer(scrubbed):
+            key = match.group(1) or match.group(2)
+            if not key:
                 continue
-            for match in self._PARAM_REF_RE.finditer(line):
-                key = match.group(1)
-                if key not in valid_keys:
-                    invalid.append(key)
+            if valid_keys and key not in valid_keys:
+                invalid.append(key)
+            elif not valid_keys and self_output_key and key == self_output_key:
+                invalid.append(key)
 
         if not invalid:
             return None
 
         unique_invalid = sorted(set(invalid))
-        return (
+        valid_list = ", ".join(repr(k) for k in sorted(valid_keys)) or "(none)"
+        message = (
             f"Invalid context.parameters references: {', '.join(repr(k) for k in unique_invalid)}. "
-            f"Valid parameter keys are: {', '.join(repr(k) for k in sorted(valid_keys))}. "
+            f"Valid parameter keys are: {valid_list}. "
             f"For fields without a matching parameter, use ai='proactive' with a descriptive prompt "
             f"instead of context.parameters['invented_name']."
         )
+        if self_output_key and self_output_key in set(invalid):
+            message += (
+                f" NOTE: {self_output_key!r} is the current block's own output slot — it is None "
+                f"during this block's execution and reading it (or any subscript of it) crashes with "
+                f"'NoneType' object is not subscriptable'. Use a parameter from the valid list, an "
+                f"upstream block's output, or ai='proactive' instead."
+            )
+        return message
 
     def _validate_parameter_preservation(
         self, new_code: str, existing_code: str | None, parameter_keys: list[str]
diff --git a/tests/unit/forge/sdk/db/test_workflow_runs_repository.py b/tests/unit/forge/sdk/db/test_workflow_runs_repository.py
index 25a09ce0c..78bd0292d 100644
--- a/tests/unit/forge/sdk/db/test_workflow_runs_repository.py
+++ b/tests/unit/forge/sdk/db/test_workflow_runs_repository.py
@@ -201,6 +201,28 @@ async def test_get_all_runs_v2_search_key_matches_run_id_and_workflow_permanent_
     assert "abc123" in where_clause
 
 
+@pytest.mark.asyncio
+async def test_get_all_runs_v2_selects_workflow_deleted_flag() -> None:
+    captured: dict[str, Any] = {}
+
+    async def _execute(query):
+        captured["query"] = query
+        return _EmptyExecuteResult()
+
+    session = MagicMock()
+    session.execute = AsyncMock(side_effect=_execute)
+
+    repo = WorkflowRunsRepository(session_factory=lambda: _SessionContext(session), debug_enabled=False)
+
+    await repo.get_all_runs_v2(organization_id="o_test")
+
+    rendered = str(captured["query"].compile(compile_kwargs={"literal_binds": True}))
+    assert "AS workflow_deleted" in rendered
+    # NOT EXISTS subquery against an active (non-deleted) workflows row.
+    assert "NOT (EXISTS" in rendered
+    assert "workflows.deleted_at IS NULL" in rendered
+
+
 @pytest.mark.asyncio
 async def test_get_all_runs_v2_excludes_copilot_session_workflow_runs() -> None:
     captured: dict[str, Any] = {}
diff --git a/tests/unit/test_runs_v2_route.py b/tests/unit/test_runs_v2_route.py
index 56a64cd90..80cae86ac 100644
--- a/tests/unit/test_runs_v2_route.py
+++ b/tests/unit/test_runs_v2_route.py
@@ -22,6 +22,7 @@ async def test_get_runs_v2_serializes_mapping_rows_from_database(monkeypatch: py
                     "finished_at": None,
                     "created_at": "2026-04-01T00:00:00Z",
                     "workflow_permanent_id": "wpid_123",
+                    "workflow_deleted": False,
                     "script_run": False,
                     "searchable_text": "Workflow run",
                 }
@@ -56,6 +57,7 @@ async def test_get_runs_v2_serializes_mapping_rows_from_database(monkeypatch: py
             "finished_at": None,
             "created_at": "2026-04-01T00:00:00Z",
             "workflow_permanent_id": "wpid_123",
+            "workflow_deleted": False,
             "script_run": False,
         }
     ]
diff --git a/tests/unit/test_script_reviewer_validators.py b/tests/unit/test_script_reviewer_validators.py
index 90e6825f1..8568bfa7c 100644
--- a/tests/unit/test_script_reviewer_validators.py
+++ b/tests/unit/test_script_reviewer_validators.py
@@ -1,8 +1,14 @@
 """Tests for ScriptReviewer validation methods."""
 
+from datetime import datetime
+from unittest.mock import AsyncMock
+
 import pytest
 
+from skyvern.forge import app
+from skyvern.schemas.scripts import ScriptFallbackEpisode
 from skyvern.services.script_reviewer import ScriptReviewer
+from tests.unit.force_stub_app import start_forge_stub_app
 
 
 class TestValidateNoHardcodedValues:
@@ -630,3 +636,472 @@ class TestLoadFilteredRunParamValues:
         ]
         result = await self._run_with_params(monkeypatch, param_tuples)
         assert result == {"real": "51410020"}
+
+
+class TestFilterSelfOutputParamKeys:
+    """Tests for _filter_self_output_param_keys.
+
+    Each block carries an auto-generated ``<block_label>_output`` parameter.
+    Inside the block's own cached function that slot is ``None`` until the
+    block has finished running, so any ``context.parameters['<self>_output']``
+    reference crashes with ``'NoneType' object is not subscriptable``.
+    Upstream and downstream block outputs are valid references — only the
+    *current* block's output must be filtered.
+    """
+
+    def test_drops_self_output_only(self) -> None:
+        keys = ["account_number", "Login_block_output", "Search_block_output"]
+        filtered = ScriptReviewer._filter_self_output_param_keys(keys, "Login_block")
+        assert filtered == ["account_number", "Search_block_output"]
+
+    def test_keeps_upstream_outputs(self) -> None:
+        """Upstream `_output` keys are legitimate cross-block references and must survive."""
+        keys = ["Search_results_output", "Login_block_output"]
+        filtered = ScriptReviewer._filter_self_output_param_keys(keys, "Login_block")
+        assert "Search_results_output" in filtered
+        assert "Login_block_output" not in filtered
+
+    def test_keeps_downstream_outputs(self) -> None:
+        """Downstream `_output` keys aren't read by the current block in practice,
+        but filtering only the self-output keeps the rule narrow and predictable."""
+        keys = ["Login_block_output", "Final_extract_output"]
+        filtered = ScriptReviewer._filter_self_output_param_keys(keys, "Login_block")
+        assert "Final_extract_output" in filtered
+        assert "Login_block_output" not in filtered
+
+    def test_empty_block_label_is_noop(self) -> None:
+        """An empty block_label means we don't know what to filter — pass through."""
+        keys = ["a", "b_output"]
+        assert ScriptReviewer._filter_self_output_param_keys(keys, "") == keys
+
+    def test_no_self_output_present(self) -> None:
+        keys = ["account_number", "username"]
+        assert ScriptReviewer._filter_self_output_param_keys(keys, "Login_block") == keys
+
+    def test_partial_match_not_filtered(self) -> None:
+        """`block_output_other` matches `<block>_output` as a prefix but isn't the
+        self-output key. The filter must be exact-match, not prefix-match."""
+        keys = ["Login_block_output_other", "Login_block_output"]
+        filtered = ScriptReviewer._filter_self_output_param_keys(keys, "Login_block")
+        assert filtered == ["Login_block_output_other"]
+
+
+class TestValidateParameterReferencesSelfOutput:
+    """Tests for _validate_parameter_references when a self-output key is in play."""
+
+    def setup_method(self) -> None:
+        self.reviewer = ScriptReviewer()
+
+    def test_self_output_reference_rejected(self) -> None:
+        """When `parameter_keys` excludes the self-output, a `context.parameters['<self>_output']`
+        reference in the LLM output must be rejected by the validator."""
+        code = """
+async def block_fn(page, context):
+    await page.fill(selector='#user', value=context.parameters['Login_block_output']['username'])
+"""
+        # parameter_keys does NOT contain Login_block_output (already filtered)
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "Login_block_output" in error
+
+    def test_self_output_error_message_explains_why(self) -> None:
+        """The retry hint should specifically explain the self-output crash so the LLM
+        doesn't reintroduce the same pattern on the next attempt."""
+        code = "value=context.parameters['Login_block_output']"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "None" in error  # message should mention the None / NoneType crash mode
+
+    def test_validator_unchanged_without_self_output_key(self) -> None:
+        """Existing callers that don't pass `self_output_key` keep the prior behavior."""
+        code = "value=context.parameters['fake_key']"
+        error = self.reviewer._validate_parameter_references(code, parameter_keys=["real_key"])
+        assert error is not None
+        assert "fake_key" in error
+        # No self-output augmentation when caller didn't opt in.
+        assert "self-output" not in error.lower()
+
+    # ------------------------------------------------------------------
+    # Empty-valid-set edge case: when the workflow's only declared parameter
+    # was the block's own output (filtered → empty parameter_keys), the
+    # validator must still scan to catch a self-output ref.
+    # ------------------------------------------------------------------
+
+    def test_self_output_rejected_with_empty_parameter_keys(self) -> None:
+        """When parameter_keys is empty BUT self_output_key is set, the
+        validator must still reject self-output references — otherwise the bug
+        slips through any workflow whose only parameter was the block's own
+        output."""
+        code = "value=context.parameters['Login_block_output']['username']"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=[],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "Login_block_output" in error
+
+    def test_no_constraints_returns_none(self) -> None:
+        """No parameter_keys AND no self_output_key → no validation possible."""
+        code = "value=context.parameters['anything']"
+        assert self.reviewer._validate_parameter_references(code, parameter_keys=[]) is None
+
+    def test_empty_valid_set_allows_unknown_non_self_refs(self) -> None:
+        """Empty ``parameter_keys`` + ``self_output_key`` in scope: the reviewer
+        rejects only the self-output ref. Other unknown refs may be runtime-
+        valid synthesized keys (``GeneratedWorkflowParameters`` field names)
+        that the reviewer doesn't load into its valid set, so we let them
+        through rather than risking false-positive rejection of legitimate
+        deterministic-named parameters.
+        """
+        code = "value=context.parameters['some_other_key']"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=[],
+            self_output_key="Login_block_output",
+        )
+        assert error is None
+
+    # ------------------------------------------------------------------
+    # `.get('key')` access: regex must catch this form too — the reviewer
+    # prompt's example output uses ``.get(...)``, so the LLM can emit it.
+    # ------------------------------------------------------------------
+
+    def test_get_access_self_output_rejected(self) -> None:
+        """`context.parameters.get('<self>_output')` must be caught the same as
+        the subscript form."""
+        code = "value=context.parameters.get('Login_block_output')"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "Login_block_output" in error
+
+    def test_get_access_with_default_self_output_rejected(self) -> None:
+        """The `.get(key, default)` two-argument form is also caught."""
+        code = "value=context.parameters.get('Login_block_output', '')"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "Login_block_output" in error
+
+    def test_get_access_invented_key_rejected(self) -> None:
+        """Generic invented-key detection also applies via the `.get()` regex."""
+        code = "value=context.parameters.get('made_up_field')"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+        )
+        assert error is not None
+        assert "made_up_field" in error
+
+    # ------------------------------------------------------------------
+    # Whitespace inside subscript brackets / `.get()` parens. Reformatted
+    # variants like ``[ 'key' ]`` must not bypass the validator.
+    # ------------------------------------------------------------------
+
+    def test_spaced_subscript_self_output_rejected(self) -> None:
+        """`context.parameters[ 'X_output' ]['username']` must be caught despite
+        whitespace inside the outer subscript brackets."""
+        code = "value=context.parameters[ 'Login_block_output' ]['username']"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "Login_block_output" in error
+
+    def test_spaced_get_access_self_output_rejected(self) -> None:
+        """`.get( 'X_output' )` form with whitespace must also be caught."""
+        code = "value=context.parameters.get( 'Login_block_output' )"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "Login_block_output" in error
+
+    # ------------------------------------------------------------------
+    # Multiline access: ``ruff format`` can wrap long subscripts across
+    # multiple lines. The validator must still match.
+    # ------------------------------------------------------------------
+
+    def test_multiline_subscript_self_output_rejected(self) -> None:
+        """A subscript split across multiple lines must still be caught."""
+        code = "value=context.parameters[\n    'Login_block_output'\n]['username']"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "Login_block_output" in error
+
+    def test_multiline_get_access_self_output_rejected(self) -> None:
+        """A `.get(...)` call split across multiple lines must still be caught."""
+        code = "value=context.parameters.get(\n    'Login_block_output',\n    {},\n)"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "Login_block_output" in error
+
+    # ------------------------------------------------------------------
+    # Whitespace between ``parameters`` and the accessor: valid Python
+    # (``context.parameters [...]``, ``context.parameters .get(...)``)
+    # must be matched too.
+    # ------------------------------------------------------------------
+
+    def test_space_before_subscript_self_output_rejected(self) -> None:
+        """`context.parameters ['X_output']` (space between `parameters` and `[`)
+        must be caught."""
+        code = "value=context.parameters ['Login_block_output']"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "Login_block_output" in error
+
+    def test_space_before_get_self_output_rejected(self) -> None:
+        """`context.parameters .get('X_output')` (space between `parameters` and
+        `.`) must be caught."""
+        code = "value=context.parameters .get('Login_block_output')"
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+            self_output_key="Login_block_output",
+        )
+        assert error is not None
+        assert "Login_block_output" in error
+
+    def test_multiline_ref_inside_comment_block_not_flagged(self) -> None:
+        """Multiline scan must still skip refs that live inside comment lines.
+        Each comment line is replaced with a blank line before scanning, so a
+        block of comment-prefixed lines containing a fake ref must not be
+        flagged as a violation."""
+        code = (
+            "# context.parameters['fake_key']\n"
+            "# context.parameters.get('another_fake_key')\n"
+            "value=context.parameters['account_number']\n"
+        )
+        error = self.reviewer._validate_parameter_references(
+            code,
+            parameter_keys=["account_number"],
+        )
+        assert error is None
+
+
+class TestConditionalCodeSelfOutputRejection:
+    """The conditional-review path also persists code via
+    ``create_script_version_from_review``, so it must run the same self-output
+    parameter validation as ``_review_block_internal``."""
+
+    def setup_method(self) -> None:
+        self.reviewer = ScriptReviewer()
+
+    @pytest.mark.asyncio
+    async def test_generate_conditional_code_calls_validator_with_self_output(self, monkeypatch: object) -> None:
+        """End-to-end wiring: drive ``_generate_conditional_code`` with a mocked
+        LLM that returns code containing the self-output pattern, and assert
+        ``_validate_parameter_references`` was invoked with the block's
+        ``self_output_key``. A future refactor that drops the validator call
+        or threads the wrong ``block_label`` will make this test fail."""
+        start_forge_stub_app()
+
+        block_label = "Branch_block"
+
+        # Conditional code containing the self-output crash pattern. The mocked
+        # LLM returns this twice (max_attempts=2) so the validator's rejection
+        # path runs both attempts.
+        bad_code = (
+            "async def block_fn(page, context):\n"
+            "    if context.parameters['Branch_block_output']['flag']:\n"
+            '        return {"next_block_label": "next", "branch_index": 0}\n'
+            "    else:\n"
+            '        return {"next_block_label": None, "branch_index": 1}\n'
+        )
+
+        async def fake_llm_handler(*args: object, **kwargs: object) -> str:
+            return f"```python\n{bad_code}\n```"
+
+        app.SCRIPT_REVIEWER_LLM_API_HANDLER = AsyncMock(side_effect=fake_llm_handler)
+
+        # Spy on the validator to confirm the wiring threads `self_output_key`.
+        validator_calls: list[dict[str, object]] = []
+        original_validator = self.reviewer._validate_parameter_references
+
+        def spy_validator(code: str, parameter_keys: list[str], self_output_key: str | None = None) -> str | None:
+            validator_calls.append(
+                {"code_len": len(code), "param_keys": list(parameter_keys), "self_output_key": self_output_key}
+            )
+            return original_validator(code, parameter_keys, self_output_key=self_output_key)
+
+        monkeypatch.setattr(self.reviewer, "_validate_parameter_references", spy_validator)
+
+        # Build a minimal episode with a branch expression so
+        # `_generate_conditional_code` proceeds past its early returns.
+        episode = ScriptFallbackEpisode(
+            episode_id="ep_1",
+            organization_id="o_test",
+            workflow_permanent_id="wpid_test",
+            workflow_run_id="wr_1",
+            block_label=block_label,
+            fallback_type="conditional_agent",
+            agent_actions={
+                "expressions": [
+                    {
+                        "original_expression": "x > 0",
+                        "rendered_expression": "1 > 0",
+                        "result": True,
+                        "is_default": False,
+                        "next_block_label": "next",
+                    },
+                    {
+                        "original_expression": None,
+                        "rendered_expression": None,
+                        "result": False,
+                        "is_default": True,
+                        "next_block_label": None,
+                    },
+                ]
+            },
+            created_at=datetime(2026, 1, 1),
+            modified_at=datetime(2026, 1, 1),
+        )
+
+        result = await self.reviewer._generate_conditional_code(
+            block_label=block_label,
+            episode=episode,
+            organization_id="o_test",
+            run_parameter_values=None,
+            all_parameter_keys=["account_number", f"{block_label}_output"],
+        )
+
+        # Validator must have been called at least once with the correct
+        # self_output_key for this block. This is the wiring assertion.
+        assert validator_calls, "validator was never invoked — wiring broken"
+        assert all(call["self_output_key"] == f"{block_label}_output" for call in validator_calls), (
+            f"validator received wrong self_output_key: {validator_calls}"
+        )
+
+        # The bad_code references the block's self-output, so the validator
+        # rejects it on every attempt and the function returns None (no code
+        # accepted for persistence). This confirms the rejection chain reaches
+        # the persistence gate.
+        assert result is None, "self-output ref should have been rejected before persistence"
+
+    @pytest.mark.asyncio
+    async def test_review_conditional_blocks_threads_parameter_keys(self, monkeypatch: object) -> None:
+        """Outer-wiring test: ``review_conditional_blocks`` loads workflow
+        context internally and threads ``all_parameter_keys`` into
+        ``_generate_conditional_code``. A refactor that drops that load (or
+        threads the wrong keys) must fail this test."""
+        start_forge_stub_app()
+
+        block_label = "Branch_block"
+
+        async def fake_llm_handler(*args: object, **kwargs: object) -> str:
+            return '```python\nasync def block_fn(page, context):\n    return {"next_block_label": None, "branch_index": 0}\n```'
+
+        app.SCRIPT_REVIEWER_LLM_API_HANDLER = AsyncMock(side_effect=fake_llm_handler)
+
+        # Spy on `_load_workflow_context` to confirm it's called from the
+        # conditional path AND to control what it returns.
+        load_calls: list[dict[str, str]] = []
+
+        async def fake_load(organization_id: str, workflow_permanent_id: str) -> tuple:
+            load_calls.append({"organization_id": organization_id, "workflow_permanent_id": workflow_permanent_id})
+            return (
+                {block_label: ""},  # goals
+                ["account_number", f"{block_label}_output"],  # all_parameter_keys
+                {},  # block_criteria
+            )
+
+        monkeypatch.setattr(self.reviewer, "_load_workflow_context", fake_load)
+
+        # Spy on the validator to confirm `_generate_conditional_code` was
+        # called WITH the keys threaded through `_load_workflow_context`.
+        validator_calls: list[dict[str, object]] = []
+        original_validator = self.reviewer._validate_parameter_references
+
+        def spy_validator(code: str, parameter_keys: list[str], self_output_key: str | None = None) -> str | None:
+            validator_calls.append({"param_keys": list(parameter_keys), "self_output_key": self_output_key})
+            return original_validator(code, parameter_keys, self_output_key=self_output_key)
+
+        monkeypatch.setattr(self.reviewer, "_validate_parameter_references", spy_validator)
+
+        episode = ScriptFallbackEpisode(
+            episode_id="ep_1",
+            organization_id="o_test",
+            workflow_permanent_id="wpid_test",
+            workflow_run_id="wr_1",
+            block_label=block_label,
+            fallback_type="conditional_agent",
+            agent_actions={
+                "expressions": [
+                    {
+                        "original_expression": "x > 0",
+                        "rendered_expression": "1 > 0",
+                        "result": True,
+                        "is_default": False,
+                        "next_block_label": None,
+                    },
+                ]
+            },
+            created_at=datetime(2026, 1, 1),
+            modified_at=datetime(2026, 1, 1),
+        )
+
+        await self.reviewer.review_conditional_blocks(
+            organization_id="o_test",
+            workflow_permanent_id="wpid_test",
+            conditional_episodes=[episode],
+            run_parameter_values=None,
+        )
+
+        # Assertion 1: workflow context was loaded by review_conditional_blocks itself.
+        assert load_calls == [{"organization_id": "o_test", "workflow_permanent_id": "wpid_test"}], (
+            "review_conditional_blocks did not load workflow context — the wiring is broken"
+        )
+        # Assertion 2: the validator received the loaded keys, with the self-output filtered out.
+        assert validator_calls, "validator was never invoked from the conditional path"
+        for call in validator_calls:
+            assert "account_number" in call["param_keys"], "upstream key should have been threaded through"
+            assert f"{block_label}_output" not in call["param_keys"], "self-output should have been filtered"
+            assert call["self_output_key"] == f"{block_label}_output", "self_output_key not threaded correctly"
+
+    def test_validator_directly_rejects_self_output(self) -> None:
+        """Direct unit-level assertion (kept alongside the integration test):
+        validator returns an error message identifying the self-output key."""
+        block_label = "Branch_block"
+        code_with_self_output = (
+            "async def block_fn(page, context):\n"
+            "    if context.parameters['Branch_block_output']['flag']:\n"
+            '        return {"next_block_label": "next", "branch_index": 0}\n'
+        )
+        param_keys = self.reviewer._filter_self_output_param_keys(sorted({"account_number"}), block_label)
+        error = self.reviewer._validate_parameter_references(
+            code_with_self_output, param_keys, self_output_key=f"{block_label}_output"
+        )
+        assert error is not None
+        assert "Branch_block_output" in error
+        assert "None" in error