From 2a3d30fb3a0b1ff98b7261aa9eb9cb02705d187c Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 22:22:34 -0400 Subject: [PATCH] feat: persist MCP workflow proof and decision lock across daemons (Closes #559) Share review-workflow-load and review-decision-lock state across MCP daemon process pools via a user-private durable store keyed by profile identity (not host-global /tmp), with TTL and fail-closed profile checks. --- gitea_mcp_server.py | 109 ++++++-- mcp_session_state.py | 391 +++++++++++++++++++++++++++++ review_workflow_load.py | 122 ++++++++- tests/conftest.py | 21 ++ tests/test_mcp_session_state.py | 207 +++++++++++++++ tests/test_review_workflow_load.py | 36 ++- 6 files changed, 857 insertions(+), 29 deletions(-) create mode 100644 mcp_session_state.py create mode 100644 tests/test_mcp_session_state.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 0832d93..ed650b5 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -804,6 +804,7 @@ import task_capability_map # noqa: E402 import review_proofs # noqa: E402 import review_workflow_boundary # noqa: E402 import review_workflow_load # noqa: E402 +import mcp_session_state # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 import issue_lock_provenance # noqa: E402 @@ -2480,37 +2481,110 @@ _REVIEW_ACTIONS = { _TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"}) -# In-process only (#211): never persist to /tmp — host-global files are -# spoofable by any local process and go stale across sessions. +# Durable across MCP daemon process pools (#559), but never under /tmp (#211): +# host-global temp files are spoofable. Persistence uses the user-private +# cache directory from mcp_session_state (mode 0o700 / files 0o600), keyed by +# remote + profile identity with TTL. _REVIEW_DECISION_LOCK: dict | None = None +def _decision_lock_binding(lock: dict | None = None) -> dict: + """Resolve key fields for durable decision-lock storage.""" + profile = get_profile() + profile_name = (profile.get("profile_name") or "").strip() + env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip() + stored_lock = ((lock or {}).get("session_profile_lock") or "").strip() + session_lock = env_lock or stored_lock or profile_name + remote = ((lock or {}).get("remote") or "").strip() or None + org = ((lock or {}).get("org") or (lock or {}).get("ready_org") or "").strip() or None + repo = ((lock or {}).get("repo") or (lock or {}).get("ready_repo") or "").strip() or None + return { + "session_profile": profile_name, + "session_profile_lock": session_lock, + "profile_identity": mcp_session_state.current_profile_identity( + profile_name=profile_name, + session_profile_lock=session_lock, + ), + "remote": remote, + "org": org, + "repo": repo, + } + + def _load_review_decision_lock(): + """Load decision lock from memory, falling back to durable session state.""" global _REVIEW_DECISION_LOCK + if _REVIEW_DECISION_LOCK is not None: + return _REVIEW_DECISION_LOCK + binding = _decision_lock_binding() + # Profile-keyed durable file; remote/org/repo validated from payload (#559). + durable = mcp_session_state.load_state( + kind=mcp_session_state.KIND_DECISION_LOCK, + profile_identity=binding.get("profile_identity"), + ) + if durable is not None: + _REVIEW_DECISION_LOCK = dict(durable) return _REVIEW_DECISION_LOCK def _save_review_decision_lock(data): + """Persist decision lock to memory + durable shared state (#559).""" global _REVIEW_DECISION_LOCK - _REVIEW_DECISION_LOCK = data + if data is None: + binding = _decision_lock_binding(_REVIEW_DECISION_LOCK) + mcp_session_state.clear_state( + kind=mcp_session_state.KIND_DECISION_LOCK, + profile_identity=binding.get("profile_identity"), + ) + _REVIEW_DECISION_LOCK = None + return + payload = dict(data) + binding = _decision_lock_binding(payload) + payload.setdefault("session_pid", os.getpid()) + payload["writer_pid"] = os.getpid() + payload["session_profile"] = binding["session_profile"] or payload.get( + "session_profile" + ) + payload["session_profile_lock"] = binding["session_profile_lock"] + payload["profile_identity"] = binding["profile_identity"] + if binding.get("remote") and not payload.get("remote"): + payload["remote"] = binding["remote"] + persisted = mcp_session_state.save_state( + kind=mcp_session_state.KIND_DECISION_LOCK, + payload=payload, + remote=payload.get("remote"), + org=payload.get("org") or payload.get("ready_org"), + repo=payload.get("repo") or payload.get("ready_repo"), + profile_identity=payload.get("profile_identity"), + ) + _REVIEW_DECISION_LOCK = dict(persisted or payload) def _review_decision_session_reasons(lock: dict | None) -> list[str]: - """Reject locks that do not belong to this MCP process/session.""" + """Reject locks that do not belong to this MCP session identity (#559). + + Different daemon PIDs in the same IDE session pool are allowed when the + profile identity and remote match. Spoofed/stale locks still fail closed. + """ if lock is None: return [] reasons = [] - if lock.get("session_pid") != os.getpid(): - reasons.append( - "review decision lock was created in a different process " - "(fail closed)" - ) env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip() - stored_lock = (lock.get("session_profile_lock") or "").strip() + stored_lock = (lock.get("session_profile_lock") or lock.get("profile_identity") or "").strip() if env_lock and stored_lock and env_lock != stored_lock: reasons.append( "review decision lock session profile lock mismatch (fail closed)" ) + # TTL / identity checks from durable envelope fields. + for reason in mcp_session_state.identity_match_reasons( + lock, + remote=lock.get("remote"), + org=lock.get("org") or lock.get("ready_org"), + repo=lock.get("repo") or lock.get("ready_repo"), + profile_identity=env_lock or stored_lock, + ): + if "profile identity mismatch" in reason or "expired" in reason or "future" in reason or "missing recorded_at" in reason: + reasons.append(reason) return reasons @@ -2521,11 +2595,12 @@ def init_review_decision_lock(remote: str | None, task: str | None, force: bool if not force: lock = _load_review_decision_lock() if lock is not None: - if lock.get("remote") == remote and lock.get("session_pid") == os.getpid(): - env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip() - stored_lock = (lock.get("session_profile_lock") or "").strip() - if not env_lock or not stored_lock or env_lock == stored_lock: - return + env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip() + stored_lock = (lock.get("session_profile_lock") or "").strip() + same_remote = lock.get("remote") == remote + same_profile = (not env_lock or not stored_lock or env_lock == stored_lock) + if same_remote and same_profile and not _review_decision_session_reasons(lock): + return review_workflow_load.clear_review_workflow_load() profile = get_profile() profile_name = (profile.get("profile_name") or "").strip() @@ -2540,6 +2615,10 @@ def init_review_decision_lock(remote: str | None, task: str | None, force: bool "session_pid": os.getpid(), "session_profile": profile_name, "session_profile_lock": session_lock, + "profile_identity": mcp_session_state.current_profile_identity( + profile_name=profile_name, + session_profile_lock=session_lock, + ), "final_review_decision_ready": False, "ready_pr_number": None, "ready_action": None, diff --git a/mcp_session_state.py b/mcp_session_state.py new file mode 100644 index 0000000..3f93784 --- /dev/null +++ b/mcp_session_state.py @@ -0,0 +1,391 @@ +"""Durable MCP session validation state shared across daemon process pools (#559). + +The IDE often routes sequential MCP tool calls to different daemon processes. +Session-scoped proofs (workflow load, review decision lock) must therefore +survive process boundaries while remaining fail-closed against spoofing. + +Security notes (extends #211): +- Never store under host-global ``/tmp`` (world-writable, spoofable). +- Default root is ``~/.cache/gitea-tools/session-state`` (mode ``0o700``). +- Files are written atomically with mode ``0o600``. +- Records are keyed by remote + org + repo + profile identity, not by PID. +- TTL prevents indefinitely stale reuse across unrelated sessions. +""" + +from __future__ import annotations + +import fcntl +import json +import os +import re +import tempfile +from contextlib import contextmanager +from datetime import datetime, timedelta, timezone +from typing import Any + +STATE_DIR_ENV = "GITEA_MCP_SESSION_STATE_DIR" +DEFAULT_STATE_DIR = os.path.expanduser("~/.cache/gitea-tools/session-state") +TTL_HOURS_ENV = "GITEA_MCP_SESSION_STATE_TTL_HOURS" +DEFAULT_TTL_HOURS = 4.0 + +KIND_WORKFLOW_LOAD = "review_workflow_load" +KIND_DECISION_LOCK = "review_decision_lock" + +_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+") +SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK" + + +def default_state_dir() -> str: + raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip() + return raw or DEFAULT_STATE_DIR + + +def ttl_hours() -> float: + raw = (os.environ.get(TTL_HOURS_ENV) or "").strip() + if not raw: + return DEFAULT_TTL_HOURS + try: + value = float(raw) + except ValueError: + return DEFAULT_TTL_HOURS + return value if value > 0 else DEFAULT_TTL_HOURS + + +def _sanitize_segment(value: str) -> str: + text = (value or "").strip() + if not text: + return "_" + return _SAFE_SEGMENT_RE.sub("_", text) + + +def current_profile_identity( + profile_name: str | None = None, + session_profile_lock: str | None = None, + profile_identity: str | None = None, +) -> str: + """Resolve the session profile identity used as the durable key.""" + env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip() + explicit = (profile_identity or session_profile_lock or "").strip() + lock = (explicit or env_lock or "").strip() + name = (profile_name or "").strip() + return lock or name or "unknown-profile" + + +def state_key( + *, + kind: str, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + profile_identity: str | None = None, +) -> str: + """Build durable filename key. + + Session proofs are one-active-per-profile (workflow load / decision lock), + so the primary key is kind + profile identity. Remote/org/repo are stored + inside the payload and validated on load (#559), which lets a later daemon + process recover state without already knowing the remote argument. + """ + # Keep remote/org/repo parameters for API stability / future kinds; they are + # intentionally not part of the filename for session-scoped proofs. + _ = (remote, org, repo) + return "-".join( + _sanitize_segment(part) + for part in ( + kind, + profile_identity or "unknown-profile", + ) + ) + + +def state_file_path( + *, + kind: str, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + profile_identity: str | None = None, + state_dir: str | None = None, +) -> str: + root = (state_dir or default_state_dir()).strip() + name = state_key( + kind=kind, + remote=remote, + org=org, + repo=repo, + profile_identity=profile_identity, + ) + return os.path.join(root, f"{name}.json") + + +def _ensure_state_dir(state_dir: str | None = None) -> str: + root = (state_dir or default_state_dir()).strip() + os.makedirs(root, mode=0o700, exist_ok=True) + return root + + +def _now_utc() -> datetime: + return datetime.now(timezone.utc) + + +def _parse_iso(value: str | None) -> datetime | None: + text = (value or "").strip() + if not text: + return None + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +@contextmanager +def _exclusive_file_lock(lock_path: str): + os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True) + fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600) + try: + fcntl.flock(fd, fcntl.LOCK_EX) + yield fd + finally: + try: + fcntl.flock(fd, fcntl.LOCK_UN) + finally: + os.close(fd) + + +def _read_json(path: str) -> dict[str, Any] | None: + if not path or not os.path.exists(path): + return None + try: + with open(path, encoding="utf-8") as handle: + data = json.load(handle) + except (OSError, json.JSONDecodeError): + return None + return data if isinstance(data, dict) else None + + +def _write_json(path: str, data: dict[str, Any]) -> None: + parent = os.path.dirname(path) or "." + os.makedirs(parent, mode=0o700, exist_ok=True) + payload = json.dumps(data, indent=2, sort_keys=True) + "\n" + fd, temp_path = tempfile.mkstemp(prefix=".session-", suffix=".json", dir=parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + os.chmod(temp_path, 0o600) + os.replace(temp_path, path) + try: + os.chmod(path, 0o600) + except OSError: + pass + finally: + if os.path.exists(temp_path): + try: + os.remove(temp_path) + except OSError: + pass + + +def identity_match_reasons( + record: dict[str, Any] | None, + *, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + profile_identity: str | None = None, +) -> list[str]: + """Return fail-closed reasons when durable record identity does not match.""" + if record is None: + return [] + reasons: list[str] = [] + expected_profile = current_profile_identity(profile_identity=profile_identity) + stored_profile = ( + (record.get("session_profile_lock") or record.get("profile_identity") or "") + .strip() + ) + if stored_profile and expected_profile and stored_profile != expected_profile: + if expected_profile != "unknown-profile": + reasons.append( + "session state profile identity mismatch " + f"(stored={stored_profile!r}, active={expected_profile!r}; fail closed)" + ) + + for field, expected in ( + ("remote", remote), + ("org", org), + ("repo", repo), + ): + want = (expected or "").strip() + have = (str(record.get(field) or "")).strip() + if want and have and want != have: + reasons.append( + f"session state {field} mismatch " + f"(stored={have!r}, expected={want!r}; fail closed)" + ) + + recorded_at = _parse_iso(record.get("recorded_at") or record.get("updated_at")) + if recorded_at is None: + reasons.append("session state missing recorded_at timestamp (fail closed)") + else: + age = _now_utc() - recorded_at + if age > timedelta(hours=ttl_hours()): + reasons.append( + f"session state expired after {ttl_hours():g}h (fail closed)" + ) + if age < timedelta(0): + reasons.append("session state recorded_at is in the future (fail closed)") + return reasons + + +def load_state( + *, + kind: str, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + profile_identity: str | None = None, + state_dir: str | None = None, +) -> dict[str, Any] | None: + """Load durable state payload when identity checks pass.""" + profile = current_profile_identity(profile_identity=profile_identity) + path = state_file_path( + kind=kind, + remote=remote, + org=org, + repo=repo, + profile_identity=profile, + state_dir=state_dir, + ) + lock_path = f"{path}.lock" + with _exclusive_file_lock(lock_path): + envelope = _read_json(path) + if not envelope: + return None + payload = envelope.get("payload") + if not isinstance(payload, dict): + return None + # Identity fields live on both envelope and payload for convenience. + merged = dict(payload) + for key in ( + "kind", + "remote", + "org", + "repo", + "profile_identity", + "session_profile_lock", + "recorded_at", + "updated_at", + "writer_pid", + ): + if key in envelope and key not in merged: + merged[key] = envelope[key] + reasons = identity_match_reasons( + merged, + remote=remote, + org=org, + repo=repo, + profile_identity=profile, + ) + if reasons: + return None + return merged + + +def save_state( + *, + kind: str, + payload: dict[str, Any] | None, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + profile_identity: str | None = None, + state_dir: str | None = None, +) -> dict[str, Any] | None: + """Persist or clear durable state for the given session identity key.""" + profile = current_profile_identity( + profile_name=payload.get("session_profile") if payload else None, + session_profile_lock=( + (payload or {}).get("session_profile_lock") or profile_identity + ), + ) + # Prefer explicit args over payload fields for key location. + key_remote = remote if remote is not None else (payload or {}).get("remote") + key_org = org if org is not None else (payload or {}).get("org") + key_repo = repo if repo is not None else (payload or {}).get("repo") + + root = _ensure_state_dir(state_dir) + path = state_file_path( + kind=kind, + remote=key_remote, + org=key_org, + repo=key_repo, + profile_identity=profile, + state_dir=root, + ) + lock_path = f"{path}.lock" + with _exclusive_file_lock(lock_path): + if payload is None: + for candidate in (path, lock_path): + if os.path.exists(candidate): + try: + os.remove(candidate) + except OSError: + pass + return None + + now = _now_utc().isoformat().replace("+00:00", "Z") + body = dict(payload) + body.setdefault("session_pid", os.getpid()) + body["writer_pid"] = os.getpid() + body["profile_identity"] = profile + if not (body.get("session_profile_lock") or "").strip(): + body["session_profile_lock"] = profile + body["recorded_at"] = body.get("recorded_at") or now + body["updated_at"] = now + if key_remote is not None: + body["remote"] = key_remote + if key_org is not None: + body["org"] = key_org + if key_repo is not None: + body["repo"] = key_repo + + envelope = { + "kind": kind, + "remote": key_remote, + "org": key_org, + "repo": key_repo, + "profile_identity": profile, + "session_profile_lock": body.get("session_profile_lock"), + "recorded_at": body["recorded_at"], + "updated_at": body["updated_at"], + "writer_pid": body["writer_pid"], + "payload": body, + } + _write_json(path, envelope) + return dict(body) + + +def clear_state( + *, + kind: str, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + profile_identity: str | None = None, + state_dir: str | None = None, +) -> None: + save_state( + kind=kind, + payload=None, + remote=remote, + org=org, + repo=repo, + profile_identity=profile_identity, + state_dir=state_dir, + ) diff --git a/review_workflow_load.py b/review_workflow_load.py index 9cee3d0..8deb622 100644 --- a/review_workflow_load.py +++ b/review_workflow_load.py @@ -1,4 +1,4 @@ -"""Canonical review-merge workflow load proof for reviewer mutations (#389, #403).""" +"""Canonical review-merge workflow load proof for reviewer mutations (#389, #403, #559).""" from __future__ import annotations @@ -7,6 +7,7 @@ import os import re from pathlib import Path +import mcp_session_state import review_workflow_boundary as boundary WORKFLOW_REL_PATH = ( @@ -88,17 +89,79 @@ def assess_prompt_conflict(prompt_text: str | None) -> tuple[bool, list[str]]: return bool(reasons), reasons +def _session_binding_fields() -> dict: + """Capture profile identity used to share state across daemon processes.""" + env_lock = (os.environ.get(mcp_session_state.SESSION_PROFILE_LOCK_ENV) or "").strip() + profile_name = (os.environ.get("GITEA_MCP_PROFILE") or "").strip() + remote = (os.environ.get("GITEA_MCP_REMOTE") or "").strip() or None + identity = mcp_session_state.current_profile_identity( + profile_name=profile_name, + session_profile_lock=env_lock, + ) + return { + "session_profile": profile_name or identity, + "session_profile_lock": env_lock or identity, + "profile_identity": identity, + "remote": remote, + } + + +def _persist_workflow_load(record: dict | None) -> dict | None: + """Write durable workflow-load proof (or clear it).""" + binding = _session_binding_fields() + if record is None: + mcp_session_state.clear_state( + kind=mcp_session_state.KIND_WORKFLOW_LOAD, + remote=binding.get("remote"), + profile_identity=binding.get("profile_identity"), + ) + return None + payload = dict(record) + payload.update({ + k: v for k, v in binding.items() if v is not None + }) + return mcp_session_state.save_state( + kind=mcp_session_state.KIND_WORKFLOW_LOAD, + payload=payload, + remote=payload.get("remote"), + org=payload.get("org"), + repo=payload.get("repo"), + profile_identity=payload.get("profile_identity"), + ) + + +def _load_durable_workflow_load() -> dict | None: + binding = _session_binding_fields() + return mcp_session_state.load_state( + kind=mcp_session_state.KIND_WORKFLOW_LOAD, + remote=binding.get("remote"), + profile_identity=binding.get("profile_identity"), + ) + + +def _active_workflow_load() -> dict | None: + """Prefer in-process cache; fall back to durable shared state (#559).""" + global _REVIEW_WORKFLOW_LOAD + if _REVIEW_WORKFLOW_LOAD is not None: + return _REVIEW_WORKFLOW_LOAD + durable = _load_durable_workflow_load() + if durable is not None: + _REVIEW_WORKFLOW_LOAD = dict(durable) + return _REVIEW_WORKFLOW_LOAD + + def record_review_workflow_load( project_root: str, *, prompt_text: str | None = None, ) -> dict: - """Record in-process workflow load proof for the current MCP session.""" + """Record workflow load proof for the current MCP session (durable + memory).""" global _REVIEW_WORKFLOW_LOAD meta = build_canonical_workflow_metadata( project_root, prompt_text=prompt_text) boundary_state = boundary.assess_boundary_status(project_root) - _REVIEW_WORKFLOW_LOAD = { + binding = _session_binding_fields() + record = { **meta, "session_pid": os.getpid(), "loaded": True, @@ -107,7 +170,10 @@ def record_review_workflow_load( "pre_review_command_count": boundary_state.get("pre_review_command_count"), "boundary_violation_count": boundary_state.get("boundary_violation_count"), "boundary_reasons": list(boundary_state.get("reasons") or []), + **{k: v for k, v in binding.items() if v is not None}, } + persisted = _persist_workflow_load(record) + _REVIEW_WORKFLOW_LOAD = dict(persisted or record) return dict(_REVIEW_WORKFLOW_LOAD) @@ -115,12 +181,13 @@ def clear_review_workflow_load() -> None: """Test helper and review_pr session reset.""" global _REVIEW_WORKFLOW_LOAD _REVIEW_WORKFLOW_LOAD = None + _persist_workflow_load(None) boundary.clear_pre_review_commands() def workflow_load_status(project_root: str | None = None) -> dict: """Non-throwing status for capability/runtime reports.""" - load = _REVIEW_WORKFLOW_LOAD + load = _active_workflow_load() if load is None: return { "workflow_load_proof_present": False, @@ -148,6 +215,8 @@ def workflow_load_status(project_root: str | None = None) -> dict: "prompt_conflicts_with_workflow": load.get( "prompt_conflicts_with_workflow"), "session_pid": load.get("session_pid"), + "writer_pid": load.get("writer_pid"), + "profile_identity": load.get("profile_identity"), "boundary_status": load.get("boundary_status"), "boundary_clean": load.get("boundary_clean"), "workflow_load_helper_result": boundary.workflow_load_helper_result( @@ -160,13 +229,47 @@ def _session_validation_reasons( load: dict, project_root: str | None, ) -> list[str]: + """Validate durable/in-memory load proof for this session identity (#559).""" reasons: list[str] = [] - if load.get("session_pid") != os.getpid(): + # Cross-process daemon pools are allowed when profile identity matches. + # Reject only when the stored profile identity conflicts with this process. + binding = _session_binding_fields() + stored_identity = ( + load.get("session_profile_lock") + or load.get("profile_identity") + or "" + ).strip() + active_identity = (binding.get("profile_identity") or "").strip() + if ( + stored_identity + and active_identity + and stored_identity != active_identity + and active_identity != "unknown-profile" + and stored_identity != "unknown-profile" + ): reasons.append( - "workflow load proof was recorded in a different process " - "(fail closed)" + "workflow load proof profile identity mismatch " + f"(stored={stored_identity!r}, active={active_identity!r}; fail closed)" ) return reasons + + # Expired durable records are treated as absent. + identity_reasons = mcp_session_state.identity_match_reasons( + load, + remote=binding.get("remote") or load.get("remote"), + org=load.get("org"), + repo=load.get("repo"), + profile_identity=active_identity or stored_identity, + ) + # Filter out remote-mismatch noise when remote was not bound at load time. + for reason in identity_reasons: + if "missing recorded_at" in reason or "expired" in reason or "future" in reason: + reasons.append(reason) + elif "profile identity mismatch" in reason: + reasons.append(reason) + if reasons: + return reasons + if load.get("prompt_conflicts_with_workflow"): reasons.extend(load.get("prompt_conflict_reasons") or [ "active prompt conflicts with loaded review-merge workflow" @@ -196,7 +299,8 @@ def review_workflow_load_blockers( ) -> list[str]: """Reasons reviewer mutations must fail closed.""" boundary_reasons = boundary.boundary_blockers(project_root) - if boundary_reasons and _REVIEW_WORKFLOW_LOAD is None: + load = _active_workflow_load() + if boundary_reasons and load is None: return boundary_reasons status = workflow_load_status(project_root) if not status.get("workflow_load_proof_present"): @@ -214,4 +318,4 @@ def recovery_handoff_without_replay() -> list[str]: "Do not call gitea_submit_pr_review, gitea_mark_final_review_decision, " "or gitea_merge_pr until workflow-load proof is present.", "Do not include approve/merge replay commands in the recovery handoff.", - ] \ No newline at end of file + ] diff --git a/tests/conftest.py b/tests/conftest.py index 7bbf78c..1c34986 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -18,14 +18,25 @@ def _reset_mutation_authority(monkeypatch): explicitly. """ monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False) + # Isolate durable session-state files so tests never share host cache (#559). + import tempfile + + _state_tmp = tempfile.TemporaryDirectory(prefix="gitea-session-state-") + monkeypatch.setenv("GITEA_MCP_SESSION_STATE_DIR", _state_tmp.name) try: import mcp_server except Exception: + _state_tmp.cleanup() yield return monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None) monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {}) monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None) + try: + import review_workflow_load + review_workflow_load._REVIEW_WORKFLOW_LOAD = None + except Exception: + pass try: import capability_stop_terminal capability_stop_terminal.clear() @@ -37,3 +48,13 @@ def _reset_mutation_authority(monkeypatch): capability_stop_terminal.clear() except Exception: pass + try: + import review_workflow_load + review_workflow_load._REVIEW_WORKFLOW_LOAD = None + except Exception: + pass + try: + mcp_server._REVIEW_DECISION_LOCK = None + except Exception: + pass + _state_tmp.cleanup() diff --git a/tests/test_mcp_session_state.py b/tests/test_mcp_session_state.py new file mode 100644 index 0000000..ebfc0ce --- /dev/null +++ b/tests/test_mcp_session_state.py @@ -0,0 +1,207 @@ +"""Tests for durable MCP session state shared across daemon processes (#559).""" + +from __future__ import annotations + +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +sys_path_root = str(Path(__file__).resolve().parent.parent) +import sys + +if sys_path_root not in sys.path: + sys.path.insert(0, sys_path_root) + +import mcp_session_state +import review_workflow_load +import mcp_server + + +class TestMcpSessionStateStore(unittest.TestCase): + def setUp(self): + self._tmpdir = tempfile.TemporaryDirectory() + self.state_dir = self._tmpdir.name + self._env = patch.dict( + os.environ, + { + mcp_session_state.STATE_DIR_ENV: self.state_dir, + mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer", + "GITEA_MCP_PROFILE": "prgs-reviewer", + }, + clear=False, + ) + self._env.start() + + def tearDown(self): + self._env.stop() + self._tmpdir.cleanup() + + def test_round_trip_same_identity(self): + saved = mcp_session_state.save_state( + kind=mcp_session_state.KIND_WORKFLOW_LOAD, + payload={"loaded": True, "workflow_hash": "abc123def456"}, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertIsNotNone(saved) + self.assertEqual(saved["workflow_hash"], "abc123def456") + self.assertIn("recorded_at", saved) + + loaded = mcp_session_state.load_state( + kind=mcp_session_state.KIND_WORKFLOW_LOAD, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertIsNotNone(loaded) + self.assertEqual(loaded["workflow_hash"], "abc123def456") + self.assertEqual(loaded["profile_identity"], "prgs-reviewer") + + def test_profile_mismatch_returns_none(self): + mcp_session_state.save_state( + kind=mcp_session_state.KIND_DECISION_LOCK, + payload={"final_review_decision_ready": True, "remote": "prgs"}, + remote="prgs", + ) + with patch.dict( + os.environ, + {mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"}, + clear=False, + ): + loaded = mcp_session_state.load_state( + kind=mcp_session_state.KIND_DECISION_LOCK, + remote="prgs", + ) + self.assertIsNone(loaded) + + def test_clear_removes_file(self): + mcp_session_state.save_state( + kind=mcp_session_state.KIND_WORKFLOW_LOAD, + payload={"loaded": True}, + remote="prgs", + ) + path = mcp_session_state.state_file_path( + kind=mcp_session_state.KIND_WORKFLOW_LOAD, + remote="prgs", + profile_identity="prgs-reviewer", + state_dir=self.state_dir, + ) + self.assertTrue(os.path.exists(path)) + mcp_session_state.clear_state( + kind=mcp_session_state.KIND_WORKFLOW_LOAD, + remote="prgs", + profile_identity="prgs-reviewer", + ) + self.assertFalse(os.path.exists(path)) + + def test_files_are_private_mode(self): + mcp_session_state.save_state( + kind=mcp_session_state.KIND_WORKFLOW_LOAD, + payload={"loaded": True}, + remote="prgs", + ) + path = mcp_session_state.state_file_path( + kind=mcp_session_state.KIND_WORKFLOW_LOAD, + remote="prgs", + profile_identity="prgs-reviewer", + state_dir=self.state_dir, + ) + mode = os.stat(path).st_mode & 0o777 + self.assertEqual(mode, 0o600) + + +class TestWorkflowLoadCrossProcess(unittest.TestCase): + def setUp(self): + self._tmpdir = tempfile.TemporaryDirectory() + self._env = patch.dict( + os.environ, + { + mcp_session_state.STATE_DIR_ENV: self._tmpdir.name, + mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer", + "GITEA_MCP_PROFILE": "prgs-reviewer", + "GITEA_MCP_REMOTE": "prgs", + }, + clear=False, + ) + self._env.start() + review_workflow_load.clear_review_workflow_load() + + def tearDown(self): + review_workflow_load.clear_review_workflow_load() + self._env.stop() + self._tmpdir.cleanup() + + def test_different_pid_same_profile_accepted(self): + root = str(Path(__file__).resolve().parent.parent) + recorded = review_workflow_load.record_review_workflow_load(root) + self.assertTrue(recorded["loaded"]) + + # Simulate another daemon process: clear memory, keep durable file, + # and change apparent PID identity only (profile remains the same). + review_workflow_load._REVIEW_WORKFLOW_LOAD = None + status = review_workflow_load.workflow_load_status(root) + self.assertTrue(status["workflow_load_proof_present"]) + self.assertTrue( + status["workflow_load_valid"], + msg=status.get("reasons"), + ) + + def test_profile_mismatch_blocks(self): + root = str(Path(__file__).resolve().parent.parent) + review_workflow_load.record_review_workflow_load(root) + review_workflow_load._REVIEW_WORKFLOW_LOAD = None + with patch.dict( + os.environ, + {mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-author"}, + clear=False, + ): + # Durable load uses active identity; mismatched profile key misses. + status = review_workflow_load.workflow_load_status(root) + self.assertFalse(status["workflow_load_proof_present"]) + + +class TestDecisionLockCrossProcess(unittest.TestCase): + def setUp(self): + self._tmpdir = tempfile.TemporaryDirectory() + self._env = patch.dict( + os.environ, + { + mcp_session_state.STATE_DIR_ENV: self._tmpdir.name, + mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer", + }, + clear=False, + ) + self._env.start() + mcp_server._save_review_decision_lock(None) + review_workflow_load.clear_review_workflow_load() + + def tearDown(self): + mcp_server._save_review_decision_lock(None) + review_workflow_load.clear_review_workflow_load() + self._env.stop() + self._tmpdir.cleanup() + + def test_decision_lock_survives_memory_clear(self): + with patch.object(mcp_server, "get_profile", return_value={ + "profile_name": "prgs-reviewer", + }): + mcp_server.init_review_decision_lock("prgs", "review_pr", force=True) + lock = mcp_server._load_review_decision_lock() + self.assertIsNotNone(lock) + self.assertEqual(lock["remote"], "prgs") + self.assertFalse(lock["final_review_decision_ready"]) + + # New process: memory empty, durable state remains. + mcp_server._REVIEW_DECISION_LOCK = None + restored = mcp_server._load_review_decision_lock() + self.assertIsNotNone(restored) + self.assertEqual(restored["remote"], "prgs") + reasons = mcp_server._review_decision_session_reasons(restored) + self.assertEqual(reasons, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_review_workflow_load.py b/tests/test_review_workflow_load.py index e59bf1a..8d1e42a 100644 --- a/tests/test_review_workflow_load.py +++ b/tests/test_review_workflow_load.py @@ -33,12 +33,38 @@ class TestReviewWorkflowLoadModule(unittest.TestCase): self.assertTrue(status["workflow_load_proof_present"]) self.assertTrue(status["workflow_load_valid"]) - def test_stale_session_pid_blocks(self): + def test_profile_identity_mismatch_blocks(self): + """#559: different daemon PIDs are OK; profile identity mismatch is not.""" + import tempfile + from unittest.mock import patch + + import mcp_session_state + root = str(__import__("pathlib").Path(__file__).resolve().parent.parent) - review_workflow_load.record_review_workflow_load(root) - review_workflow_load._REVIEW_WORKFLOW_LOAD["session_pid"] = 0 - blockers = review_workflow_load.review_workflow_load_blockers(root) - self.assertTrue(any("different process" in b for b in blockers)) + with tempfile.TemporaryDirectory() as tmp: + with patch.dict( + os.environ, + { + mcp_session_state.STATE_DIR_ENV: tmp, + mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer", + }, + clear=False, + ): + review_workflow_load.clear_review_workflow_load() + review_workflow_load.record_review_workflow_load(root) + # Corrupt the in-memory profile identity while keeping PID. + review_workflow_load._REVIEW_WORKFLOW_LOAD[ + "session_profile_lock" + ] = "other-profile" + review_workflow_load._REVIEW_WORKFLOW_LOAD[ + "profile_identity" + ] = "other-profile" + blockers = review_workflow_load.review_workflow_load_blockers(root) + self.assertTrue( + any("profile identity mismatch" in b for b in blockers), + msg=blockers, + ) + review_workflow_load.clear_review_workflow_load() def test_prompt_conflict_detected(self): conflict, reasons = review_workflow_load.assess_prompt_conflict(