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

@@ -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,

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,
+    )

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.",
-    ]
+    ]

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()

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()

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(