sysadmin
d302602567
Add shared workflow_scope_guard with typed blockers (blocker_kind +
exact_next_action) and force-on production enforcement under pytest.
Wire root/branches/scope checks through verify_preflight_purity and
mutation entrypoints (create_issue, comment_issue) without adopting the
rejected 300a4ca patterns (test-mode early-return, porcelain *.py filter).
Regression suite covers out-of-scope issue ownership, root diagnostic
edits, worktree bind success path, force-on under pytest, porcelain
integrity, monkeypatch resistance, real entrypoint fail-closed proof,
and durable failure recording.
627 lines
22 KiB
Python
627 lines
22 KiB
Python
"""Workflow scope ownership and production-guard hardening (#683).
|
|
|
|
Implements fail-closed enforcement so sessions cannot:
|
|
|
|
* mutate source/tests on the root/control checkout (including temporary
|
|
diagnostic edits) without binding an issue-backed ``branches/`` worktree;
|
|
* continue out-of-scope source work while locked to a different issue;
|
|
* disable, skip, or conceal production root/branches/porcelain guards solely
|
|
because pytest/unittest is loaded.
|
|
|
|
This module is pure assessment + small durable ledger helpers. Callers gather
|
|
live facts (lock, branch, porcelain, worktree path) and pass them in. Existing
|
|
root_checkout_guard / author_mutation_worktree assessors remain authoritative;
|
|
this module composes typed blockers with exact recovery actions.
|
|
|
|
Do **not** reintroduce the rejected #681 / ``300a4ca`` patterns:
|
|
|
|
* early-return from workspace verification under ``_preflight_in_test_mode()``
|
|
* porcelain filtering that strips ``*.py`` lines under pytest
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import threading
|
|
from typing import Any
|
|
|
|
import author_mutation_worktree
|
|
from reviewer_worktree import parse_dirty_tracked_files
|
|
|
|
# ── force-on / test isolation ────────────────────────────────────────────────
|
|
|
|
# When set, production root/branches/scope guards MUST run even under pytest.
|
|
# Unit tests that only need preflight-order isolation leave this unset and
|
|
# use GITEA_TEST_PORCELAIN / fixtures; real-entrypoint proof sets this to "1".
|
|
FORCE_PRODUCTION_GUARDS_ENV = "GITEA_TEST_FORCE_PRODUCTION_GUARDS"
|
|
|
|
# Existing force signals also mean "exercise production dirtiness paths".
|
|
_FORCE_DIRTY_ENV = "GITEA_TEST_FORCE_DIRTY"
|
|
_FORCE_PORCELAIN_ENV = "GITEA_TEST_PORCELAIN"
|
|
|
|
# ── typed blocker kinds ──────────────────────────────────────────────────────
|
|
|
|
BLOCKER_ROOT_DIAGNOSTIC_EDIT = "root_diagnostic_edit"
|
|
BLOCKER_MISSING_ISSUE_SCOPE = "missing_issue_scope"
|
|
BLOCKER_OUT_OF_SCOPE_ISSUE = "out_of_scope_issue"
|
|
BLOCKER_MISSING_WORKTREE = "missing_issue_worktree"
|
|
BLOCKER_UNRECORDED_FAILURE = "unrecorded_workflow_failure"
|
|
BLOCKER_PRODUCTION_GUARD = "production_guard_violation"
|
|
|
|
BLOCKER_KINDS = frozenset(
|
|
{
|
|
BLOCKER_ROOT_DIAGNOSTIC_EDIT,
|
|
BLOCKER_MISSING_ISSUE_SCOPE,
|
|
BLOCKER_OUT_OF_SCOPE_ISSUE,
|
|
BLOCKER_MISSING_WORKTREE,
|
|
BLOCKER_UNRECORDED_FAILURE,
|
|
BLOCKER_PRODUCTION_GUARD,
|
|
}
|
|
)
|
|
|
|
_NEXT_ACTIONS: dict[str, str] = {
|
|
BLOCKER_ROOT_DIAGNOSTIC_EDIT: (
|
|
"Stop editing the control/root checkout. Preserve or discard root WIP "
|
|
"durably, restore root to clean master, lock or create the owning issue, "
|
|
"bind branches/issue-<N>-*, set GITEA_AUTHOR_WORKTREE to that worktree, "
|
|
"then re-run the mutation."
|
|
),
|
|
BLOCKER_MISSING_ISSUE_SCOPE: (
|
|
"Select or create the owning Gitea issue, claim/lock it "
|
|
"(gitea_mark_issue + gitea_lock_issue), bind branches/issue-<N>-* "
|
|
"from clean master, then re-run the mutation from that worktree."
|
|
),
|
|
BLOCKER_OUT_OF_SCOPE_ISSUE: (
|
|
"Stop. The active issue lock does not own this work. Release or finish "
|
|
"the current issue lease, then select/create and lock the correct "
|
|
"owning issue, bind its branches/issue-<N>-* worktree, and re-run."
|
|
),
|
|
BLOCKER_MISSING_WORKTREE: (
|
|
"Bind an issue-backed worktree under branches/ (scripts/worktree-start "
|
|
"or git worktree add branches/issue-<N>-*), set GITEA_AUTHOR_WORKTREE / "
|
|
"worktree_path to that path, keep the control checkout clean on master, "
|
|
"then re-run the mutation."
|
|
),
|
|
BLOCKER_UNRECORDED_FAILURE: (
|
|
"Record the workflow/tool failure durably first (issue comment or "
|
|
"workflow_scope_guard.record_workflow_failure), then continue only "
|
|
"inside the owning issue-backed worktree."
|
|
),
|
|
BLOCKER_PRODUCTION_GUARD: (
|
|
"Resolve the production guard violation: clean or isolate the control "
|
|
"checkout, bind the owning issue worktree under branches/, and re-run "
|
|
"with production guards active."
|
|
),
|
|
}
|
|
|
|
_ISSUE_IN_BRANCH_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
|
|
|
|
# In-process durable failure ledger (also written via optional sink callback).
|
|
_ledger_lock = threading.Lock()
|
|
_failure_ledger: list[dict[str, Any]] = []
|
|
|
|
|
|
class ProductionGuardError(RuntimeError):
|
|
"""Fail-closed production guard with typed blocker metadata (#683)."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
*,
|
|
blocker_kind: str,
|
|
exact_next_action: str | None = None,
|
|
reasons: list[str] | None = None,
|
|
details: dict[str, Any] | None = None,
|
|
) -> None:
|
|
super().__init__(message)
|
|
kind = (blocker_kind or "").strip()
|
|
if kind not in BLOCKER_KINDS:
|
|
kind = BLOCKER_PRODUCTION_GUARD
|
|
self.blocker_kind = kind
|
|
self.exact_next_action = (
|
|
(exact_next_action or "").strip() or _NEXT_ACTIONS[kind]
|
|
)
|
|
self.reasons = list(reasons or [message])
|
|
self.details = dict(details or {})
|
|
|
|
|
|
def production_guards_forced() -> bool:
|
|
"""True when the explicit #683 force-on flag requests production guards."""
|
|
return (os.environ.get(FORCE_PRODUCTION_GUARDS_ENV) or "").strip().lower() in {
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
"on",
|
|
}
|
|
|
|
|
|
def purity_order_forced() -> bool:
|
|
"""True when tests force preflight-order dirtiness paths (legacy flags)."""
|
|
if os.environ.get(_FORCE_DIRTY_ENV):
|
|
return True
|
|
# GITEA_TEST_PORCELAIN present (even empty) means dirtiness paths are live.
|
|
if os.environ.get(_FORCE_PORCELAIN_ENV) is not None:
|
|
return True
|
|
return False
|
|
|
|
|
|
def production_guards_active(*, in_test_mode: bool) -> bool:
|
|
"""Whether production root/branches/scope guards must execute.
|
|
|
|
Production (non-test) always active. Under pytest, active when either the
|
|
explicit #683 force-on flag or legacy dirty/porcelain force signals are
|
|
set — never skip production enforcement solely because tests are running
|
|
when force-on is requested.
|
|
"""
|
|
if production_guards_forced() or purity_order_forced():
|
|
return True
|
|
return not bool(in_test_mode)
|
|
|
|
|
|
def extract_issue_number_from_branch(branch_name: str | None) -> int | None:
|
|
"""Return the first issue-N number embedded in a branch name, if any."""
|
|
text = (branch_name or "").strip()
|
|
if not text:
|
|
return None
|
|
match = _ISSUE_IN_BRANCH_RE.search(text)
|
|
if not match:
|
|
return None
|
|
try:
|
|
return int(match.group(1))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def is_source_or_test_path(path: str) -> bool:
|
|
"""True for tracked source/test paths that must not land as root WIP."""
|
|
p = (path or "").replace("\\", "/").lstrip("./")
|
|
if not p:
|
|
return False
|
|
if p.startswith("tests/") or "/tests/" in f"/{p}":
|
|
return True
|
|
if p.endswith((".py", ".pyi", ".toml", ".cfg", ".ini", ".sh")):
|
|
return True
|
|
if p in {"requirements.txt", "pyproject.toml", "setup.py", "setup.cfg"}:
|
|
return True
|
|
return False
|
|
|
|
|
|
def dirty_source_files(porcelain_status: str) -> list[str]:
|
|
"""Tracked dirty paths that count as source/test contamination."""
|
|
dirty = parse_dirty_tracked_files(porcelain_status or "")
|
|
return [p for p in dirty if is_source_or_test_path(p)]
|
|
|
|
|
|
def assess_issue_scope_ownership(
|
|
*,
|
|
locked_issue_number: int | None,
|
|
target_issue_number: int | None = None,
|
|
branch_name: str | None = None,
|
|
role_kind: str | None = None,
|
|
require_lock_for_author: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Fail closed when the session issue lock does not own the attempted work.
|
|
|
|
* Author sessions that require a lock fail when none is held.
|
|
* When a lock exists, the target issue (tool argument) and/or the issue
|
|
number embedded in the branch must match the locked issue.
|
|
* Reviewer/merger/reconciler roles are not issue-scope owners of author
|
|
implementation work and skip the author lock requirement.
|
|
"""
|
|
role = (role_kind or "").strip().lower()
|
|
locked = locked_issue_number
|
|
if isinstance(locked, str) and locked.isdigit():
|
|
locked = int(locked)
|
|
if locked is not None:
|
|
try:
|
|
locked = int(locked)
|
|
except (TypeError, ValueError):
|
|
locked = None
|
|
|
|
target = target_issue_number
|
|
if target is not None:
|
|
try:
|
|
target = int(target)
|
|
except (TypeError, ValueError):
|
|
target = None
|
|
|
|
branch_issue = extract_issue_number_from_branch(branch_name)
|
|
reasons: list[str] = []
|
|
blocker_kind: str | None = None
|
|
|
|
# Non-author roles do not take author issue locks for implementation.
|
|
if role in {"reviewer", "merger", "reconciler"}:
|
|
return _scope_ok(locked, target, branch_issue)
|
|
|
|
if require_lock_for_author and locked is None:
|
|
reasons.append(
|
|
"no owning issue lock is bound for this author session; "
|
|
"source/test mutation requires selecting or creating an owning issue first"
|
|
)
|
|
blocker_kind = BLOCKER_MISSING_ISSUE_SCOPE
|
|
|
|
if locked is not None and target is not None and locked != target:
|
|
reasons.append(
|
|
f"session is locked to issue #{locked} but mutation targets issue "
|
|
f"#{target}; out-of-scope until the owning issue is selected"
|
|
)
|
|
blocker_kind = BLOCKER_OUT_OF_SCOPE_ISSUE
|
|
|
|
if locked is not None and branch_issue is not None and locked != branch_issue:
|
|
reasons.append(
|
|
f"session is locked to issue #{locked} but workspace branch is for "
|
|
f"issue #{branch_issue}; bind the matching issue-backed worktree"
|
|
)
|
|
blocker_kind = BLOCKER_OUT_OF_SCOPE_ISSUE
|
|
|
|
if reasons:
|
|
kind = blocker_kind or BLOCKER_MISSING_ISSUE_SCOPE
|
|
return {
|
|
"proven": False,
|
|
"block": True,
|
|
"blocker_kind": kind,
|
|
"exact_next_action": _NEXT_ACTIONS[kind],
|
|
"reasons": reasons,
|
|
"locked_issue_number": locked,
|
|
"target_issue_number": target,
|
|
"branch_issue_number": branch_issue,
|
|
}
|
|
return _scope_ok(locked, target, branch_issue)
|
|
|
|
|
|
def assess_root_source_mutation(
|
|
*,
|
|
workspace_path: str,
|
|
canonical_repo_root: str,
|
|
porcelain_status: str,
|
|
current_branch: str | None = None,
|
|
locked_issue_number: int | None = None,
|
|
role_kind: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Fail closed for diagnostic/source edits on the control/root checkout.
|
|
|
|
Allowed only when the active workspace is under ``branches/``. Dirty
|
|
tracked source/test files on the control checkout always block, including
|
|
temporary/diagnostic/test-only intent.
|
|
"""
|
|
role = (role_kind or "").strip().lower()
|
|
if role == "reconciler":
|
|
return {
|
|
"proven": True,
|
|
"block": False,
|
|
"blocker_kind": None,
|
|
"exact_next_action": "proceed",
|
|
"reasons": [],
|
|
"dirty_source_files": [],
|
|
}
|
|
|
|
root = os.path.realpath(canonical_repo_root or "")
|
|
workspace = os.path.realpath(workspace_path or root or ".")
|
|
under_branches = author_mutation_worktree.is_path_under_branches(workspace, root)
|
|
dirty_src = dirty_source_files(porcelain_status)
|
|
reasons: list[str] = []
|
|
blocker_kind: str | None = None
|
|
|
|
if not under_branches and workspace == root and dirty_src:
|
|
# Root workspace with source dirtiness is unattributed root WIP.
|
|
# (Clean-root author binding is enforced by branches-only #274.)
|
|
reasons.append(
|
|
"control/root checkout has tracked source or test edits "
|
|
f"(dirty files: {', '.join(dirty_src)}); diagnostic or temporary "
|
|
"edits on the root checkout are forbidden"
|
|
)
|
|
blocker_kind = BLOCKER_ROOT_DIAGNOSTIC_EDIT
|
|
|
|
if (
|
|
not under_branches
|
|
and workspace == root
|
|
and not dirty_src
|
|
and role == "author"
|
|
):
|
|
# Explicit missing-worktree signal for force-on author entrypoints.
|
|
reasons.append(
|
|
"author source/test mutation from the stable control checkout is "
|
|
"forbidden; bind an issue-backed worktree under branches/ first"
|
|
)
|
|
blocker_kind = BLOCKER_MISSING_WORKTREE
|
|
|
|
if reasons:
|
|
kind = blocker_kind or BLOCKER_ROOT_DIAGNOSTIC_EDIT
|
|
return {
|
|
"proven": False,
|
|
"block": True,
|
|
"blocker_kind": kind,
|
|
"exact_next_action": _NEXT_ACTIONS[kind],
|
|
"reasons": reasons,
|
|
"dirty_source_files": dirty_src,
|
|
"workspace_path": workspace,
|
|
"canonical_repo_root": root,
|
|
"under_branches": under_branches,
|
|
"locked_issue_number": locked_issue_number,
|
|
}
|
|
return {
|
|
"proven": True,
|
|
"block": False,
|
|
"blocker_kind": None,
|
|
"exact_next_action": "proceed",
|
|
"reasons": [],
|
|
"dirty_source_files": dirty_src,
|
|
"workspace_path": workspace,
|
|
"canonical_repo_root": root,
|
|
"under_branches": under_branches,
|
|
"locked_issue_number": locked_issue_number,
|
|
}
|
|
|
|
|
|
def assess_production_mutation_guards(
|
|
*,
|
|
workspace_path: str,
|
|
canonical_repo_root: str,
|
|
porcelain_status: str,
|
|
current_branch: str | None = None,
|
|
locked_issue_number: int | None = None,
|
|
target_issue_number: int | None = None,
|
|
role_kind: str | None = None,
|
|
require_author_lock: bool = False,
|
|
in_test_mode: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Compose root + scope production guards when they must be active (#683)."""
|
|
if not production_guards_active(in_test_mode=in_test_mode):
|
|
return {
|
|
"proven": True,
|
|
"block": False,
|
|
"blocker_kind": None,
|
|
"exact_next_action": "proceed",
|
|
"reasons": [],
|
|
"skipped": True,
|
|
"skip_reason": "production guards not active (test isolation without force-on)",
|
|
}
|
|
|
|
root_assess = assess_root_source_mutation(
|
|
workspace_path=workspace_path,
|
|
canonical_repo_root=canonical_repo_root,
|
|
porcelain_status=porcelain_status,
|
|
current_branch=current_branch,
|
|
locked_issue_number=locked_issue_number,
|
|
role_kind=role_kind,
|
|
)
|
|
if root_assess["block"]:
|
|
return {**root_assess, "skipped": False}
|
|
|
|
scope_assess = assess_issue_scope_ownership(
|
|
locked_issue_number=locked_issue_number,
|
|
target_issue_number=target_issue_number,
|
|
branch_name=current_branch,
|
|
role_kind=role_kind,
|
|
require_lock_for_author=require_author_lock,
|
|
)
|
|
if scope_assess["block"]:
|
|
return {**scope_assess, "skipped": False}
|
|
|
|
return {
|
|
"proven": True,
|
|
"block": False,
|
|
"blocker_kind": None,
|
|
"exact_next_action": "proceed",
|
|
"reasons": [],
|
|
"skipped": False,
|
|
"root": root_assess,
|
|
"scope": scope_assess,
|
|
}
|
|
|
|
|
|
def raise_if_blocked(assessment: dict[str, Any]) -> None:
|
|
"""Raise :class:`ProductionGuardError` when *assessment* blocks."""
|
|
if not assessment or not assessment.get("block"):
|
|
return
|
|
kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD
|
|
reasons = list(assessment.get("reasons") or ["production guard violation"])
|
|
message = (
|
|
f"Workflow scope guard (#683) [{kind}]: {'; '.join(reasons)}. "
|
|
f"exact_next_action: {assessment.get('exact_next_action') or _NEXT_ACTIONS.get(kind, '')}"
|
|
)
|
|
raise ProductionGuardError(
|
|
message,
|
|
blocker_kind=kind,
|
|
exact_next_action=assessment.get("exact_next_action"),
|
|
reasons=reasons,
|
|
details={
|
|
k: v
|
|
for k, v in assessment.items()
|
|
if k
|
|
not in {
|
|
"proven",
|
|
"block",
|
|
"blocker_kind",
|
|
"exact_next_action",
|
|
"reasons",
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
def block_response(
|
|
assessment: dict[str, Any] | ProductionGuardError | None = None,
|
|
*,
|
|
blocker_kind: str | None = None,
|
|
reasons: list[str] | None = None,
|
|
exact_next_action: str | None = None,
|
|
**extra: Any,
|
|
) -> dict[str, Any]:
|
|
"""Structured fail-closed tool response with typed blocker fields."""
|
|
if isinstance(assessment, ProductionGuardError):
|
|
kind = assessment.blocker_kind
|
|
reason_list = list(assessment.reasons)
|
|
next_action = assessment.exact_next_action
|
|
extra = {**assessment.details, **extra}
|
|
elif isinstance(assessment, dict) and assessment.get("block"):
|
|
kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD
|
|
reason_list = list(assessment.get("reasons") or [])
|
|
next_action = assessment.get("exact_next_action") or _NEXT_ACTIONS.get(
|
|
kind, _NEXT_ACTIONS[BLOCKER_PRODUCTION_GUARD]
|
|
)
|
|
else:
|
|
kind = (blocker_kind or BLOCKER_PRODUCTION_GUARD).strip()
|
|
if kind not in BLOCKER_KINDS:
|
|
kind = BLOCKER_PRODUCTION_GUARD
|
|
reason_list = list(reasons or ["production guard violation"])
|
|
next_action = exact_next_action or _NEXT_ACTIONS[kind]
|
|
|
|
if kind not in BLOCKER_KINDS:
|
|
kind = BLOCKER_PRODUCTION_GUARD
|
|
if not reason_list:
|
|
reason_list = ["production guard violation"]
|
|
next_action = (next_action or "").strip() or _NEXT_ACTIONS[kind]
|
|
|
|
out: dict[str, Any] = {
|
|
"success": False,
|
|
"performed": False,
|
|
"blocker_kind": kind,
|
|
"exact_next_action": next_action,
|
|
"reasons": reason_list,
|
|
}
|
|
for key, value in extra.items():
|
|
if key not in out and value is not None:
|
|
out[key] = value
|
|
return out
|
|
|
|
|
|
def format_production_guard_error(assessment: dict[str, Any]) -> str:
|
|
"""Single RuntimeError string carrying kind + exact next action."""
|
|
kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD
|
|
reasons = "; ".join(assessment.get("reasons") or ["production guard violation"])
|
|
next_action = assessment.get("exact_next_action") or _NEXT_ACTIONS.get(
|
|
kind, _NEXT_ACTIONS[BLOCKER_PRODUCTION_GUARD]
|
|
)
|
|
return (
|
|
f"Workflow scope guard (#683) [{kind}]: {reasons}. "
|
|
f"exact_next_action: {next_action}"
|
|
)
|
|
|
|
|
|
# ── durable failure recording ────────────────────────────────────────────────
|
|
|
|
|
|
def record_workflow_failure(
|
|
*,
|
|
kind: str,
|
|
detail: str,
|
|
issue_number: int | None = None,
|
|
task: str | None = None,
|
|
sink: Any | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Record a workflow/tool failure before source edits continue (#683 AC8).
|
|
|
|
*sink* may be a callable ``sink(record)`` (e.g. tests) or omitted for the
|
|
in-process ledger only. Returns the durable record.
|
|
"""
|
|
record = {
|
|
"kind": (kind or "workflow_failure").strip() or "workflow_failure",
|
|
"detail": (detail or "").strip(),
|
|
"issue_number": issue_number,
|
|
"task": task,
|
|
"pid": os.getpid(),
|
|
}
|
|
with _ledger_lock:
|
|
_failure_ledger.append(dict(record))
|
|
if callable(sink):
|
|
sink(record)
|
|
return record
|
|
|
|
|
|
def clear_workflow_failure_ledger() -> None:
|
|
"""Test helper: reset the in-process failure ledger."""
|
|
with _ledger_lock:
|
|
_failure_ledger.clear()
|
|
|
|
|
|
def workflow_failure_ledger() -> list[dict[str, Any]]:
|
|
"""Copy of durable in-process failure records."""
|
|
with _ledger_lock:
|
|
return [dict(r) for r in _failure_ledger]
|
|
|
|
|
|
def assess_durable_failure_recorded(
|
|
*,
|
|
require_record: bool,
|
|
pending_source_mutation: bool,
|
|
) -> dict[str, Any]:
|
|
"""Block source mutation when a workflow failure was not recorded first."""
|
|
if not require_record or not pending_source_mutation:
|
|
return {
|
|
"proven": True,
|
|
"block": False,
|
|
"blocker_kind": None,
|
|
"exact_next_action": "proceed",
|
|
"reasons": [],
|
|
}
|
|
with _ledger_lock:
|
|
has_record = bool(_failure_ledger)
|
|
if has_record:
|
|
return {
|
|
"proven": True,
|
|
"block": False,
|
|
"blocker_kind": None,
|
|
"exact_next_action": "proceed",
|
|
"reasons": [],
|
|
}
|
|
return {
|
|
"proven": False,
|
|
"block": True,
|
|
"blocker_kind": BLOCKER_UNRECORDED_FAILURE,
|
|
"exact_next_action": _NEXT_ACTIONS[BLOCKER_UNRECORDED_FAILURE],
|
|
"reasons": [
|
|
"workflow/tool failure triggered a need for source changes but no "
|
|
"durable failure record exists yet"
|
|
],
|
|
}
|
|
|
|
|
|
def porcelain_preserves_python_paths(porcelain_status: str) -> bool:
|
|
"""Regression helper: dirty ``*.py`` lines must remain visible (#683)."""
|
|
text = porcelain_status or ""
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
if stripped.endswith(".py") or ".py " in stripped or stripped.endswith(".py"):
|
|
# Any py path present proves no silent strip of all *.py lines.
|
|
if " M " in f" {stripped}" or stripped[:1] in "MADRCTU" or len(line) >= 4:
|
|
return True
|
|
# Empty porcelain is fine; integrity means we did not strip when present.
|
|
return ".py" not in text
|
|
|
|
|
|
def assert_no_pytest_porcelain_filter(source_text: str) -> list[str]:
|
|
"""Static check: production reader must not strip ``*.py`` under pytest."""
|
|
findings: list[str] = []
|
|
lowered = source_text or ""
|
|
if "endswith(\".py\")" in lowered or "endswith('.py')" in lowered:
|
|
if "pytest" in lowered and "porcelain" in lowered.lower():
|
|
findings.append(
|
|
"production porcelain reader must not filter *.py under pytest "
|
|
"(rejected 300a4ca pattern)"
|
|
)
|
|
if "if \"pytest\" in sys.modules" in lowered and "porcelain" in lowered.lower():
|
|
if ".py" in lowered and ("join" in lowered or "endswith" in lowered):
|
|
findings.append(
|
|
"test-mode porcelain filtering of source files is forbidden (#683)"
|
|
)
|
|
return findings
|
|
|
|
|
|
def _scope_ok(
|
|
locked: int | None,
|
|
target: int | None,
|
|
branch_issue: int | None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"proven": True,
|
|
"block": False,
|
|
"blocker_kind": None,
|
|
"exact_next_action": "proceed",
|
|
"reasons": [],
|
|
"locked_issue_number": locked,
|
|
"target_issue_number": target,
|
|
"branch_issue_number": branch_issue,
|
|
}
|