fix(mcp): seed cross-repo session identity from configured canonical root (#706 F1)

Addresses REQUEST_CHANGES review 457 on PR #736.

Root cause: _trusted_session_repository derived the session org/repository
solely from _workspace_repository_slug -> _local_git_remote_url, which runs
`git remote get-url` in PROJECT_ROOT (the Gitea-Tools install checkout). A
cross-repository namespace with a configured canonical_repository_root then
pinned the Gitea-Tools identity while #274 filesystem membership bound the
external target, so _enforce_canonical_repository_root failed closed on a
self-inflicted identity mismatch and the mcp-control-plane / eagenda
namespaces stayed blocked end-to-end.

Fix: when a canonical_repository_root is configured (env over profile) and
passes existence / git-toplevel / resolvable-remote-identity validation, the
session repository slug is derived from repository_identity_slug(configured
root) instead of the install remote. The derived identity is still authorized
by the profile allowed_repositories allowlist (never self-authorizing); the
canonical filesystem root and org/repo identity remain immutable for the
session; invalid / non-git / unresolvable / unallowlisted / forged / drift
cases fail closed; unconfigured single-repo Gitea-Tools namespaces keep the
install-derived slug unchanged.

Tests: new tests/test_issue_706_f1_seed_identity_integration.py drives the
real _seed_session_context -> _enforce_canonical_repository_root path with two
real temporary git repositories (install=Gitea-Tools, configured target=
mcp-control-plane): env + profile config, reviewer + merger role kinds, and
fail-closed cases (nonexistent / non-git / unallowlisted / forged identity
drift). Reproduces the F1 block before the fix; green after.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

canonical_repository_root.py

@@ -0,0 +1,267 @@
+"""Immutable canonical repository root for cross-repository namespaces (#706).
+
+The Gitea-Tools MCP server historically derived the ``canonical_repo_root`` from
+the *install checkout* the server script lives in
+(``PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))``). A namespace
+that runs the same server script against an *external* repository (e.g.
+``eagenda-author`` targeting ``eAgenda``) then failed every mutation: the
+branches-only / worktree-membership guards (#274) compared the task workspace
+against the Gitea-Tools ``.git`` directory, which it can never belong to.
+
+This module separates two distinct concepts:
+
+* the immutable code/install root (``PROJECT_ROOT``) — where the server lives, and
+* the namespace-scoped **canonical repository root** — the working root of the
+  repository whose issues/PRs the namespace mutates.
+
+The canonical repository root is configured per namespace (profile field or an
+environment variable, typically set alongside the namespace ``cwd`` in the MCP
+config). It is validated (existence, git identity, git common-directory
+membership) and pinned immutably into the session context so a later call cannot
+forge or swap it. When *no* binding is configured the single-repo default is
+preserved unchanged: the canonical root is derived from the process checkout.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from typing import Mapping
+
+import remote_repo_guard
+
+# Namespace-scoped override, typically exported next to the server ``cwd`` in the
+# MCP config for a cross-repository namespace.
+CANONICAL_ROOT_ENV = "GITEA_CANONICAL_REPOSITORY_ROOT"
+
+# Candidate git remote names probed when deriving repository identity.
+_IDENTITY_REMOTE_CANDIDATES = ("prgs", "origin", "dadeschools", "mdcps")
+
+
+def configured_canonical_root(
+    profile: Mapping | None,
+    env: Mapping | None,
+) -> tuple[str | None, str | None]:
+    """Return ``(value, source)`` for the declared canonical repository root.
+
+    Precedence: the ``GITEA_CANONICAL_REPOSITORY_ROOT`` environment variable
+    (namespace-scoped) overrides the profile ``canonical_repository_root``
+    field. Blank values are treated as unset. Returns ``(None, None)`` when no
+    binding is declared (the single-repo default).
+    """
+    env_map = env if env is not None else os.environ
+    env_val = (env_map.get(CANONICAL_ROOT_ENV) or "").strip()
+    if env_val:
+        return env_val, f"{CANONICAL_ROOT_ENV} environment variable"
+    if profile:
+        prof_val = (profile.get("canonical_repository_root") or "").strip()
+        if prof_val:
+            return prof_val, "profile canonical_repository_root"
+    return None, None
+
+
+def resolve_repo_toplevel(path: str) -> str | None:
+    """Realpath of the git working-tree top level for *path*, or None."""
+    text = (path or "").strip()
+    if not text:
+        return None
+    try:
+        res = subprocess.run(
+            ["git", "-C", text, "rev-parse", "--show-toplevel"],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+    except Exception:
+        return None
+    top = (res.stdout or "").strip()
+    return os.path.realpath(top) if top else None
+
+
+def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None:
+    """``owner/repository`` derived from a git remote configured at *path*.
+
+    Tries the caller-named remote first, then a small set of known remote names,
+    then whatever remote the repository actually has. Returns None when no remote
+    URL is parseable (identity cannot be proven).
+    """
+    text = (path or "").strip()
+    if not text:
+        return None
+
+    ordered: list[str] = []
+    for name in (remote, *_IDENTITY_REMOTE_CANDIDATES):
+        clean = (name or "").strip()
+        if clean and clean not in ordered:
+            ordered.append(clean)
+
+    try:
+        listed = subprocess.run(
+            ["git", "-C", text, "remote"],
+            capture_output=True,
+            text=True,
+            check=True,
+        ).stdout.split()
+    except Exception:
+        listed = []
+    for name in listed:
+        if name and name not in ordered:
+            ordered.append(name)
+
+    for name in ordered:
+        try:
+            url = subprocess.run(
+                ["git", "-C", text, "remote", "get-url", name],
+                capture_output=True,
+                text=True,
+                check=True,
+            ).stdout.strip()
+        except Exception:
+            continue
+        parsed = remote_repo_guard.parse_org_repo_from_remote_url(url)
+        if parsed:
+            return f"{parsed[0]}/{parsed[1]}"
+    return None
+
+
+def assess_canonical_repository_root(
+    *,
+    configured_value: str | None,
+    source: str | None,
+    expected_slug: str | None,
+    process_project_root: str,
+    remote: str | None = None,
+    require_binding: bool = False,
+) -> dict:
+    """Validate the canonical repository root binding, failing closed on forgery.
+
+    Returns a dict with ``proven`` / ``block`` / ``reasons`` plus the resolved
+    ``canonical_repo_root`` (the value downstream guards must use),
+    ``configured`` (whether a cross-repo binding was declared),
+    ``resolved_slug`` and ``source``.
+
+    Without a configured binding the single-repo default is preserved: the
+    canonical root is derived from *process_project_root* and never blocks
+    (unless *require_binding* explicitly demands one).
+
+    With a configured binding the path must exist, be a git repository, and —
+    when *expected_slug* is known — carry a matching repository identity. A
+    mismatched or (when *require_binding*) unprovable identity is a forged or
+    conflicting binding and fails closed.
+    """
+    process_root = os.path.realpath(process_project_root)
+    declared = (configured_value or "").strip()
+
+    if not declared:
+        if require_binding:
+            return _assessment(
+                proven=False,
+                reasons=[
+                    "no canonical_repository_root configured for a cross-repository "
+                    f"namespace; set {CANONICAL_ROOT_ENV} or the profile "
+                    "canonical_repository_root field (fail closed)"
+                ],
+                configured=False,
+                canonical_repo_root=process_root,
+                resolved_slug=None,
+                source=None,
+            )
+        # Single-repo default: canonical root follows the install checkout.
+        derived = resolve_repo_toplevel(process_root) or process_root
+        return _assessment(
+            proven=True,
+            reasons=[],
+            configured=False,
+            canonical_repo_root=derived,
+            resolved_slug=None,
+            source=None,
+        )
+
+    real = os.path.realpath(os.path.abspath(declared))
+    if not os.path.isdir(real):
+        return _assessment(
+            proven=False,
+            reasons=[
+                f"configured canonical repository root '{real}' does not exist "
+                "or is not a directory (fail closed)"
+            ],
+            configured=True,
+            canonical_repo_root=real,
+            resolved_slug=None,
+            source=source,
+        )
+
+    toplevel = resolve_repo_toplevel(real)
+    if not toplevel:
+        return _assessment(
+            proven=False,
+            reasons=[
+                f"configured canonical repository root '{real}' is not a git "
+                "repository (fail closed)"
+            ],
+            configured=True,
+            canonical_repo_root=real,
+            resolved_slug=None,
+            source=source,
+        )
+
+    resolved_slug = repository_identity_slug(toplevel, remote=remote)
+    reasons: list[str] = []
+    expected = (expected_slug or "").strip() or None
+    if expected:
+        if resolved_slug and resolved_slug.lower() != expected.lower():
+            reasons.append(
+                f"canonical repository root identity mismatch: '{toplevel}' resolves "
+                f"to repository '{resolved_slug}' but the session is authorized for "
+                f"'{expected}' (forged or conflicting binding, fail closed)"
+            )
+        elif not resolved_slug and require_binding:
+            reasons.append(
+                f"canonical repository root '{toplevel}' has no resolvable git "
+                f"remote identity to confirm authorization for '{expected}' "
+                "(fail closed)"
+            )
+
+    return _assessment(
+        proven=not reasons,
+        reasons=reasons,
+        configured=True,
+        canonical_repo_root=toplevel,
+        resolved_slug=resolved_slug,
+        source=source,
+    )
+
+
+def format_canonical_repository_root_error(assessment: Mapping) -> str:
+    """Single RuntimeError message for MCP preflight gates."""
+    root = assessment.get("canonical_repo_root") or "(unknown)"
+    source = assessment.get("source") or "(unconfigured)"
+    reasons = "; ".join(
+        assessment.get("reasons") or ["unknown canonical repository root violation"]
+    )
+    return (
+        f"Canonical repository root guard (#706): {reasons}. "
+        f"binding source: {source}; canonical repository root: {root}. "
+        "Configure a valid canonical_repository_root for the target repository "
+        "and relaunch; do not point it at the Gitea-Tools install checkout."
+    )
+
+
+def _assessment(
+    *,
+    proven: bool,
+    reasons: list[str],
+    configured: bool,
+    canonical_repo_root: str,
+    resolved_slug: str | None,
+    source: str | None,
+) -> dict:
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": list(reasons),
+        "configured": configured,
+        "canonical_repo_root": canonical_repo_root,
+        "resolved_slug": resolved_slug,
+        "source": source,
+    }

gitea_auth.py

@@ -802,6 +802,10 @@ def get_profile():
         # environment variable must never widen or forge the set of
         # repositories a session may bind to.
         "allowed_repositories": _json_list("allowed_repositories"),
+        # #706 cross-repository canonical root binding. Config-sourced here (the
+        # namespace-scoped GITEA_CANONICAL_REPOSITORY_ROOT env override is applied
+        # by canonical_repository_root.configured_canonical_root, not widened here).
+        "canonical_repository_root": jp.get("canonical_repository_root") or None,
         "audit_label": audit_label,
         "token_source_name": token_source,
         "auth_source_type": auth_type,

gitea_config.py

@@ -502,6 +502,30 @@ def _validate_allowed_repositories(name, raw):
             )
 
 
+def _validate_canonical_repository_root(name, raw):
+    """Validate the optional per-profile canonical repository root (#706).
+
+    ``canonical_repository_root`` binds a cross-repository namespace to the
+    working root of its target repository (separate from the immutable
+    Gitea-Tools install checkout). It is an absolute filesystem path; existence
+    and git identity are validated at bind time by the runtime guard, not here
+    (config validation stays filesystem-independent). Absent means the
+    single-repo default and is allowed.
+    """
+    if raw is None:
+        return
+    if not isinstance(raw, str) or not raw.strip():
+        raise ConfigError(
+            f"profile '{name}' canonical_repository_root must be a non-empty "
+            "absolute path string to the target repository working root"
+        )
+    if not os.path.isabs(raw.strip()):
+        raise ConfigError(
+            f"profile '{name}' canonical_repository_root {raw!r} must be an "
+            "absolute path"
+        )
+
+
 def _reject_inline_secrets(kind, name, obj):
     for key in _INLINE_SECRET_KEYS:
         if key in obj:
@@ -577,6 +601,9 @@ def _load_v2_contexts(data, path):
         if not isinstance(allowed, list) or not isinstance(forbidden, list):
             raise ConfigError(f"profile '{name}' operation fields must be lists")
         _validate_allowed_repositories(name, raw.get("allowed_repositories"))
+        _validate_canonical_repository_root(
+            name, raw.get("canonical_repository_root")
+        )
         allowed_n = {_normalize_op("gitea", op, name) for op in allowed}
         forbidden_n = {_normalize_op("gitea", op, name) for op in forbidden}
         # Reviewer-identity deadlock rule (#100/#103) applies here unchanged.

gitea_mcp_server.py

@@ -194,6 +194,7 @@ MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
 RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
 
 import namespace_workspace_binding as nwb  # noqa: E402
+import canonical_repository_root as crr  # noqa: E402  # #706 cross-repo canonical root
 import mcp_namespace_health  # noqa: E402
 import stale_binding_recovery  # noqa: E402
 
@@ -376,6 +377,22 @@ def _assess_stale_active_binding(auto_recover: bool = False) -> dict:
     return report
 
 
+def _configured_canonical_root() -> tuple[str | None, str | None]:
+    """Declared cross-repository canonical root for the active namespace (#706).
+
+    Returns ``(value, source)`` from the profile/env binding, or ``(None, None)``
+    for the single-repo default. The raw declared value is threaded into
+    workspace resolution so the branches-only / membership guards evaluate
+    against the configured target repository; the strict identity/existence
+    checks run separately in :func:`_enforce_canonical_repository_root`.
+    """
+    try:
+        profile = get_profile()
+    except Exception:
+        profile = {}
+    return crr.configured_canonical_root(profile, os.environ)
+
+
 def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
     """Resolve the namespace-scoped workspace root inspected by pre-flight guards.
 
@@ -399,8 +416,9 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
 
 
 def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict:
-    """Canonical namespace workspace + repository root for guards (#460/#510)."""
+    """Canonical namespace workspace + repository root for guards (#460/#510/#706)."""
     role = _effective_workspace_role()
+    configured_root, _source = _configured_canonical_root()
     return nwb.resolve_namespace_mutation_context(
         role_kind=role,
         worktree_path=worktree_path,
@@ -409,6 +427,7 @@ def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dic
             _reviewer_session_worktree() if role in {"reviewer", "merger"} else None
         ),
         profile_name=get_profile().get("profile_name"),
+        configured_canonical_root=configured_root,
     )
 
 
@@ -721,6 +740,54 @@ def record_preflight_check(
             _preflight_resolved_task = resolved_task
 
 
+def _enforce_canonical_repository_root(
+    worktree_path: str | None = None,
+    *,
+    remote: str | None = None,
+) -> None:
+    """#706: validate the immutable cross-repository canonical root binding.
+
+    A no-op for the single-repo default (no configured binding). When a
+    namespace declares a ``canonical_repository_root`` (profile/env), the
+    configured target repository must exist, be a git repository, and carry a
+    repository identity matching the session's authorized repository. Missing,
+    conflicting, or forged bindings fail closed, and the validated root is
+    checked against the immutable session pin so it cannot be swapped mid
+    session.
+    """
+    configured_value, source = _configured_canonical_root()
+    if not configured_value:
+        return
+
+    bound = session_ctx.get_session_context() or {}
+    expected_slug = session_ctx.format_repository_slug(
+        bound.get("org"), bound.get("repository")
+    )
+    assessment = crr.assess_canonical_repository_root(
+        configured_value=configured_value,
+        source=source,
+        expected_slug=expected_slug,
+        process_project_root=PROJECT_ROOT,
+        remote=remote,
+        require_binding=True,
+    )
+    if assessment["block"]:
+        raise RuntimeError(
+            crr.format_canonical_repository_root_error(assessment)
+        )
+
+    drift = session_ctx.assess_session_context(
+        profile_name=get_profile().get("profile_name"),
+        remote=remote,
+        canonical_repository_root=assessment["canonical_repo_root"],
+    )
+    if drift["block"]:
+        raise RuntimeError(
+            "Canonical repository root guard (#706): "
+            + "; ".join(drift["reasons"])
+        )
+
+
 def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
     """#274: author file/branch mutations must run from a branches/ worktree.
 
@@ -767,6 +834,15 @@ def _anti_stomp_in_test_mode() -> bool:
     return not bool(os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP"))
 
 
+# Mutation tasks that carry an explicit cross-repository *target* contract and
+# must NOT auto-resolve the workspace repository for omitted coordinates. Only
+# ``delete_branch`` (#733) qualifies: a destructive raw-branch deletion names
+# its org/repo explicitly and fails closed otherwise. Every other mutation is
+# session-bound to the single workspace-aligned repository and resolves it from
+# the trusted git remote (mirroring :func:`_resolve`).
+_EXPLICIT_TARGET_MUTATION_TASKS = frozenset({"delete_branch"})
+
+
 def _run_anti_stomp_preflight(
     task: str | None,
     *,
@@ -832,21 +908,56 @@ def _run_anti_stomp_preflight(
     local_remote_url = None
     org_explicit = org is not None
     repo_explicit = repo is not None
+    filled_org = False
+    filled_repo = False
     if remote:
         try:
             local_remote_url = _local_git_remote_url(remote)
         except Exception:
             local_remote_url = None
         if resolved_org is None or resolved_repo is None:
-            try:
-                rem = _effective_remote(remote)
-                if rem in REMOTES:
-                    if resolved_org is None:
-                        resolved_org = REMOTES[rem]["org"]
-                    if resolved_repo is None:
-                        resolved_repo = REMOTES[rem]["repo"]
-            except Exception:
-                pass
+            # Session-bound mutations act on the single workspace-aligned
+            # repository, so prefer the trusted git-remote-derived identity for
+            # omitted coordinates — exactly as :func:`_resolve` does — instead
+            # of the remote-wide ``REMOTES`` default *target*. A bare ``prgs``
+            # default resolves to ``Timesheet`` and false-positived the #530
+            # repo guard against a ``Gitea-Tools`` checkout, blocking native
+            # issue/PR/review/merge/lock/cleanup mutations at preflight while
+            # the actual resolve stage (``_resolve``) would have targeted the
+            # workspace repo correctly. Workspace-filled sides are intentional
+            # alignment and are marked explicit so the guard accepts them.
+            # ``delete_branch`` keeps its explicit cross-repo target contract
+            # (#733) and is excluded, so it still fails closed on omitted
+            # coordinates. ``REMOTES`` stays a best-effort last resort only when
+            # no workspace identity can be derived (preserves the dadeschools
+            # default-target behavior and the existing unit suite).
+            workspace_parts = None
+            if task not in _EXPLICIT_TARGET_MUTATION_TASKS:
+                try:
+                    workspace_parts = (
+                        remote_repo_guard.parse_org_repo_from_remote_url(
+                            local_remote_url
+                        )
+                    )
+                except Exception:
+                    workspace_parts = None
+            if workspace_parts:
+                if resolved_org is None:
+                    resolved_org = workspace_parts[0]
+                    filled_org = True
+                if resolved_repo is None:
+                    resolved_repo = workspace_parts[1]
+                    filled_repo = True
+            if resolved_org is None or resolved_repo is None:
+                try:
+                    rem = _effective_remote(remote)
+                    if rem in REMOTES:
+                        if resolved_org is None:
+                            resolved_org = REMOTES[rem]["org"]
+                        if resolved_repo is None:
+                            resolved_repo = REMOTES[rem]["repo"]
+                except Exception:
+                    pass
 
     parity = _current_master_parity()
     startup_head = parity.get("startup_head")
@@ -900,8 +1011,10 @@ def _run_anti_stomp_preflight(
         resolved_org=resolved_org,
         resolved_repo=resolved_repo,
         local_remote_url=local_remote_url,
-        org_explicit=org_explicit,
-        repo_explicit=repo_explicit,
+        # Workspace-filled sides are intentional alignment for #530 (same
+        # contract as :func:`_resolve`), so treat them as explicit.
+        org_explicit=org_explicit or filled_org,
+        repo_explicit=repo_explicit or filled_repo,
         check_repo=check_repo,
         profile_name=profile_name,
         profile_role=profile_role,
@@ -1069,6 +1182,7 @@ def verify_preflight_purity(
                     )
 
         # Historical path: root + branches after purity-order when dirty paths live.
+        _enforce_canonical_repository_root(worktree_path, remote=remote)
         _enforce_root_checkout_guard(worktree_path)
         _enforce_branches_only_author_mutation(worktree_path)
         _enforce_issue_scope_guard(
@@ -1102,6 +1216,7 @@ def verify_preflight_purity(
     # #683: under pytest unit isolation, FORCE_PRODUCTION_GUARDS still runs
     # production root + branches + issue scope (no silent no-op of guards).
     if production_active:
+        _enforce_canonical_repository_root(worktree_path, remote=remote)
         _enforce_root_checkout_guard(worktree_path)
         _enforce_branches_only_author_mutation(worktree_path)
         _enforce_issue_scope_guard(
@@ -1560,7 +1675,43 @@ def _trusted_session_repository(
             "repository": None,
             "reasons": [str(exc)],
         }
-    slug = _workspace_repository_slug(remote)
+    # #706 F1: when a cross-repository canonical root is configured, the session
+    # repository identity must be derived from that validated *target* repository,
+    # not from the install-checkout git remote. ``_workspace_repository_slug``
+    # reads ``_local_git_remote_url`` in ``PROJECT_ROOT`` (always Gitea-Tools), so
+    # without this a cross-repo namespace pinned Gitea-Tools while #274 filesystem
+    # membership bound the external root, and ``_enforce_canonical_repository_root``
+    # then failed closed on a self-inflicted identity mismatch. The derived
+    # identity is still authorized by the profile allowlist below (never
+    # self-authorizing). Env-over-profile precedence and fail-closed validation
+    # (existence, git toplevel, resolvable remote identity) are handled by
+    # ``crr``; unconfigured single-repo namespaces keep the install-derived slug.
+    configured_value, configured_source = crr.configured_canonical_root(
+        profile, os.environ
+    )
+    if configured_value:
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=configured_value,
+            source=configured_source,
+            expected_slug=None,
+            process_project_root=PROJECT_ROOT,
+            remote=remote,
+            require_binding=True,
+        )
+        slug = assessment.get("resolved_slug")
+        if assessment.get("block") or not slug:
+            return {
+                "org": None,
+                "repository": None,
+                "reasons": assessment.get("reasons")
+                or [
+                    "configured canonical repository root has no resolvable "
+                    "repository identity; session repository cannot be "
+                    "authorized (fail closed)"
+                ],
+            }
+    else:
+        slug = _workspace_repository_slug(remote)
     scope = session_ctx.assess_repository_scope(
         workspace_slug=slug,
         allowed=allowed,
@@ -1598,6 +1749,18 @@ def _seed_session_context(
     """
     expected = (profile.get("username") or "").strip() or None
     trusted = _trusted_session_repository(profile, remote)
+    # #706: pin the configured cross-repository canonical root immutably so a
+    # later call cannot forge/swap it. Store the resolved toplevel so drift
+    # checks compare against the same value the mutation gate validates.
+    configured_value, _crr_source = crr.configured_canonical_root(
+        profile, os.environ
+    )
+    canonical_root_pin = None
+    if configured_value:
+        canonical_root_pin = (
+            crr.resolve_repo_toplevel(configured_value)
+            or os.path.realpath(configured_value)
+        )
     return session_ctx.seed_session_context_if_unbound(
         profile_name=profile.get("profile_name") or "",
         remote=remote,
@@ -1608,6 +1771,7 @@ def _seed_session_context(
         role_kind=_profile_role_kind(profile),
         expected_username=expected,
         source=source,
+        canonical_repository_root=canonical_root_pin,
     )
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
@@ -2696,8 +2860,13 @@ def gitea_create_issue(
         return blocked
     try:
         with _mutation_stage("preflight_purity"):
+            # #735: forward explicit org/repo into shared anti-stomp preflight.
             verify_preflight_purity(
-                remote, worktree_path=worktree_path, task="create_issue"
+                remote,
+                worktree_path=worktree_path,
+                task="create_issue",
+                org=org,
+                repo=repo,
             )
     except Exception as exc:
         typed = _production_guard_block_from_exc(exc, number=None)
@@ -2919,7 +3088,14 @@ def gitea_lock_issue(
         )
     else:
         git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
-    verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
+    # #735: forward explicit org/repo into shared anti-stomp preflight.
+    verify_preflight_purity(
+        remote,
+        worktree_path=resolved_worktree,
+        task="lock_issue",
+        org=org,
+        repo=repo,
+    )
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -3150,7 +3326,14 @@ def gitea_create_pr(
     if blocked:
         return blocked
     with _mutation_stage("preflight_purity"):
-        verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
+        # #735: forward explicit org/repo into shared anti-stomp preflight.
+        verify_preflight_purity(
+            remote,
+            worktree_path=worktree_path,
+            task="create_pr",
+            org=org,
+            repo=repo,
+        )
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
@@ -7446,9 +7629,9 @@ def gitea_edit_pr(
     # Closing uses close_pr; non-closing title/body/base edits use edit_pr so
     # the shared #604 anti-stomp inventory stays consistent with wiring.
     if closing:
-        verify_preflight_purity(remote, task="close_pr")
+        verify_preflight_purity(remote, task="close_pr", org=org, repo=repo)
     else:
-        verify_preflight_purity(remote, task="edit_pr")
+        verify_preflight_purity(remote, task="edit_pr", org=org, repo=repo)
 
     # PR closure is a first-class capability, distinct from retitling or
     # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
@@ -7699,7 +7882,8 @@ def gitea_commit_files(
             branch="",
         )
 
-    verify_preflight_purity(remote, task="commit_files")
+    # #735: forward explicit org/repo into shared anti-stomp preflight.
+    verify_preflight_purity(remote, task="commit_files", org=org, repo=repo)
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
     h, o, r = _resolve(remote, host, org, repo)
@@ -8814,6 +8998,8 @@ def gitea_cleanup_merged_pr_branch(
         remote,
         worktree_path=worktree_path,
         task="cleanup_merged_pr_branch",
+        org=org,
+        repo=repo,
     )
 
     h, o, r = _resolve(remote, host, org, repo)
@@ -9541,7 +9727,9 @@ def gitea_reconcile_merged_cleanups(
         report["executed"] = False
         return {"success": True, "performed": False, **report}
 
-    verify_preflight_purity(remote, task="reconcile_merged_cleanups")
+    verify_preflight_purity(
+        remote, task="reconcile_merged_cleanups", org=org, repo=repo
+    )
     actions: list[dict] = []
     for entry in report.get("entries") or []:
         head_branch = entry.get("head_branch") or ""
@@ -10070,7 +10258,9 @@ def gitea_reconcile_already_landed_pr(
         )
         return result
 
-    verify_preflight_purity(remote, task="reconcile_already_landed_pr")
+    verify_preflight_purity(
+        remote, task="reconcile_already_landed_pr", org=org, repo=repo
+    )
 
     if post_comment and comment_body.strip():
         comment_block = _profile_operation_gate("gitea.pr.comment")
@@ -10341,7 +10531,9 @@ def gitea_reconcile_superseded_by_merged_pr(
         result["safe_next_action"] = "rerun with explicit mutation flags"
         return result
 
-    verify_preflight_purity(remote, task="reconcile_close_superseded_pr")
+    verify_preflight_purity(
+        remote, task="reconcile_close_superseded_pr", org=org, repo=repo
+    )
 
     if post_comment:
         comment_url = f"{base}/issues/{target_pr_number}/comments"
@@ -10444,7 +10636,7 @@ def gitea_close_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="close_issue")
+    verify_preflight_purity(remote, task="close_issue", org=org, repo=repo)
 
     # #529: block closure that buries an unproven non-zero suite exit as an
     # "expected pre-existing failure". Skipped when no closure report is given.
@@ -12411,10 +12603,13 @@ def gitea_create_issue_comment(
         # allowed while an author holds a different implementation lock.
         # Scope ownership for source edits is enforced via branch/worktree
         # binding + root diagnostic checks (#683).
+        # #735: forward explicit org/repo into shared anti-stomp preflight.
         verify_preflight_purity(
             remote,
             worktree_path=worktree_path,
             task="comment_issue",
+            org=org,
+            repo=repo,
         )
     except Exception as exc:
         typed = _production_guard_block_from_exc(
@@ -14550,7 +14745,14 @@ def gitea_mark_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
+    # #735: forward explicit org/repo into shared anti-stomp preflight.
+    verify_preflight_purity(
+        remote,
+        worktree_path=worktree_path,
+        task="mark_issue",
+        org=org,
+        repo=repo,
+    )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -14640,7 +14842,7 @@ def gitea_post_heartbeat(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="post_heartbeat")
+    verify_preflight_purity(remote, task="post_heartbeat", org=org, repo=repo)
     active_profile = profile or get_profile().get("profile_name")
     body = issue_claim_heartbeat.format_heartbeat_body(
         kind="progress",
@@ -15719,7 +15921,9 @@ def gitea_cleanup_stale_claims(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="cleanup_stale_claims")
+    verify_preflight_purity(
+        remote, task="cleanup_stale_claims", org=org, repo=repo
+    )
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -15839,7 +16043,13 @@ def gitea_create_label(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_label")
+    verify_preflight_purity(
+        remote,
+        worktree_path=worktree_path,
+        task="create_label",
+        org=org,
+        repo=repo,
+    )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -15893,7 +16103,13 @@ def gitea_set_issue_labels(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="set_issue_labels")
+    verify_preflight_purity(
+        remote,
+        worktree_path=worktree_path,
+        task="set_issue_labels",
+        org=org,
+        repo=repo,
+    )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)

namespace_workspace_binding.py

@@ -109,8 +109,17 @@ def resolve_namespace_mutation_context(
     session_lease_worktree: str | None = None,
     worktree: str | None = None,
     profile_name: str | None = None,
+    configured_canonical_root: str | None = None,
 ) -> dict:
-    """Shared workspace resolution for runtime_context and mutation guards."""
+    """Shared workspace resolution for runtime_context and mutation guards.
+
+    When *configured_canonical_root* is supplied (a cross-repository namespace
+    bound to an external target repository, #706), the canonical repository root
+    is that configured target rather than the MCP install checkout. This keeps
+    the branches-only / worktree-membership guards (#274) evaluating against the
+    repository the namespace actually mutates. Without it the single-repo
+    default is preserved: the canonical root follows the process checkout.
+    """
     demotions: list[str] = []
     workspace, binding_source = resolve_namespace_workspace(
         role_kind=role_kind,
@@ -132,7 +141,11 @@ def resolve_namespace_mutation_context(
         env=env,
         profile_name=profile_name,
     )
-    canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
+    configured = (configured_canonical_root or "").strip()
+    if configured:
+        canonical_root = os.path.realpath(configured)
+    else:
+        canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
     return {
         "workspace_path": workspace,
         "workspace_binding_source": binding_source,
@@ -258,6 +271,7 @@ def assess_namespace_mutation_workspace(
     session_lease_worktree: str | None = None,
     profile_name: str | None = None,
     current_branch: str | None = None,
+    configured_canonical_root: str | None = None,
 ) -> dict:
     """Evaluate namespace workspace binding before preflight/mutation."""
     ctx = resolve_namespace_mutation_context(
@@ -268,6 +282,7 @@ def assess_namespace_mutation_workspace(
         env=env,
         session_lease_worktree=session_lease_worktree,
         profile_name=profile_name,
+        configured_canonical_root=configured_canonical_root,
     )
     mutation_workspace = ctx["workspace_path"]
     binding_source = ctx["workspace_binding_source"]

session_context_binding.py

@@ -32,6 +32,7 @@ class _SessionContext:
     expected_username: str | None
     source: str
     pid: int
+    canonical_repository_root: str | None = None
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -45,6 +46,7 @@ class _SessionContext:
             "expected_username": self.expected_username,
             "source": self.source,
             "pid": self.pid,
+            "canonical_repository_root": self.canonical_repository_root,
         }
 
 
@@ -143,6 +145,7 @@ def bind_session_context(
     role_kind: str | None = None,
     expected_username: str | None = None,
     source: str = "bind",
+    canonical_repository_root: str | None = None,
 ) -> dict[str, Any]:
     """Atomically bind/re-bind context (the explicit activation path)."""
     with _SESSION_CONTEXT_LOCK:
@@ -156,6 +159,7 @@ def bind_session_context(
             role_kind=role_kind,
             expected_username=expected_username,
             source=source,
+            canonical_repository_root=canonical_repository_root,
         )
 
 
@@ -170,6 +174,7 @@ def _bind_session_context_unlocked(
     role_kind: str | None,
     expected_username: str | None,
     source: str,
+    canonical_repository_root: str | None = None,
 ) -> dict[str, Any]:
     """Store a complete immutable context while the caller holds the lock."""
     global _SESSION_CONTEXT
@@ -184,6 +189,7 @@ def _bind_session_context_unlocked(
         expected_username=(expected_username or "").strip() or None,
         source=source,
         pid=os.getpid(),
+        canonical_repository_root=(canonical_repository_root or "").strip() or None,
     )
     return _SESSION_CONTEXT.as_dict()
 
@@ -199,6 +205,7 @@ def seed_session_context_if_unbound(
     role_kind: str | None = None,
     expected_username: str | None = None,
     source: str = "seed",
+    canonical_repository_root: str | None = None,
 ) -> dict[str, Any]:
     """Atomically bind only when this process has no current context.
 
@@ -219,6 +226,7 @@ def seed_session_context_if_unbound(
                 role_kind=role_kind,
                 expected_username=expected_username,
                 source=source,
+                canonical_repository_root=canonical_repository_root,
             )
         return _SESSION_CONTEXT.as_dict()
 
@@ -232,6 +240,7 @@ def assess_session_context(
     repository: str | None = None,
     org: str | None = None,
     expected_username: str | None = None,
+    canonical_repository_root: str | None = None,
     require_bound: bool = False,
     require_complete: bool = False,
 ) -> dict[str, Any]:
@@ -272,6 +281,7 @@ def assess_session_context(
     live_identity = (identity or "").strip() or None
     live_repo = (repository or "").strip() or None
     live_org = (org or "").strip() or None
+    live_canonical = (canonical_repository_root or "").strip() or None
 
     if ctx.get("profile_name") and live_profile and live_profile != ctx.get("profile_name"):
         reasons.append(
@@ -307,6 +317,13 @@ def assess_session_context(
             f"org drift: live '{live_org}' != bound "
             f"'{ctx.get('org')}' (fail closed)"
         )
+    bound_canonical = ctx.get("canonical_repository_root")
+    if bound_canonical and live_canonical and live_canonical != bound_canonical:
+        reasons.append(
+            f"canonical repository root drift: live '{live_canonical}' != bound "
+            f"'{bound_canonical}' (forged or conflicting cross-repository "
+            "binding, fail closed)"
+        )
 
     expected = expected_username or ctx.get("expected_username")
     if expected and live_identity and live_identity != expected:
@@ -580,6 +597,7 @@ def mutation_context_audit_fields(
         "session_org": data.get("org"),
         "session_role_kind": data.get("role_kind"),
         "session_context_source": data.get("source"),
+        "session_canonical_repository_root": data.get("canonical_repository_root"),
     }
 
 

tests/test_issue_706_canonical_repository_root.py

@@ -0,0 +1,423 @@
+"""Issue #706: immutable startup/profile canonical_repository_root capability.
+
+Cross-repository MCP namespaces (e.g. ``eagenda-author``) must bind their
+canonical repository root to the *configured target repository*, not to the
+Gitea-Tools install checkout the server script lives in. These tests prove:
+
+* the configured binding is resolved from profile/env with env precedence,
+* the target repository identity and git common-directory membership are
+  validated (AC3),
+* the branches-only guard (#274) is enforced *inside the target repo* (AC4),
+* missing / conflicting / forged bindings fail closed (AC5),
+* prgs and mdcps namespaces stay simultaneously isolated (AC6),
+* the session binding pins the canonical root immutably,
+* no weakening of the install-root single-repo default (AC8).
+
+Uses two distinct real git repositories/worktrees (AC7).
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import canonical_repository_root as crr  # noqa: E402
+import gitea_config  # noqa: E402
+import namespace_workspace_binding as nwb  # noqa: E402
+import session_context_binding as session_ctx  # noqa: E402
+
+
+def _git(cwd: str, *args: str) -> str:
+    res = subprocess.run(
+        ["git", "-C", cwd, *args],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    return res.stdout.strip()
+
+
+def _init_repo(path: Path, remote_url: str, *, remote_name: str = "origin") -> str:
+    """Create a real git repo with one commit and a remote; return realpath root."""
+    path.mkdir(parents=True, exist_ok=True)
+    _git(str(path), "init", "-q")
+    _git(str(path), "config", "user.email", "[email protected]")
+    _git(str(path), "config", "user.name", "Test")
+    _git(str(path), "remote", "add", remote_name, remote_url)
+    (path / "README.md").write_text("seed\n")
+    _git(str(path), "add", "README.md")
+    _git(str(path), "commit", "-q", "-m", "seed")
+    return os.path.realpath(str(path))
+
+
+def _add_worktree(repo_root: str, worktree_path: Path, branch: str) -> str:
+    _git(repo_root, "worktree", "add", "-q", "-b", branch, str(worktree_path))
+    return os.path.realpath(str(worktree_path))
+
+
+# ---------------------------------------------------------------------------
+# configured_canonical_root: resolution + precedence
+# ---------------------------------------------------------------------------
+class TestConfiguredCanonicalRoot(unittest.TestCase):
+    def test_unconfigured_returns_none(self):
+        value, source = crr.configured_canonical_root({}, {})
+        self.assertIsNone(value)
+        self.assertIsNone(source)
+
+    def test_profile_field_used(self):
+        value, source = crr.configured_canonical_root(
+            {"canonical_repository_root": "/repo/eAgenda"}, {}
+        )
+        self.assertEqual(value, "/repo/eAgenda")
+        self.assertIn("profile", source)
+
+    def test_env_overrides_profile(self):
+        value, source = crr.configured_canonical_root(
+            {"canonical_repository_root": "/repo/eAgenda"},
+            {crr.CANONICAL_ROOT_ENV: "/repo/other"},
+        )
+        self.assertEqual(value, "/repo/other")
+        self.assertIn(crr.CANONICAL_ROOT_ENV, source)
+
+    def test_blank_values_ignored(self):
+        value, source = crr.configured_canonical_root(
+            {"canonical_repository_root": "   "},
+            {crr.CANONICAL_ROOT_ENV: ""},
+        )
+        self.assertIsNone(value)
+        self.assertIsNone(source)
+
+
+# ---------------------------------------------------------------------------
+# assess_canonical_repository_root: default (single-repo) path
+# ---------------------------------------------------------------------------
+class TestUnconfiguredFallback(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = self._tmp.name
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_unconfigured_falls_back_to_process_root(self):
+        root = _init_repo(
+            Path(self.tmp) / "install",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
+        )
+        got = crr.assess_canonical_repository_root(
+            configured_value=None,
+            source=None,
+            expected_slug=None,
+            process_project_root=root,
+        )
+        self.assertTrue(got["proven"])
+        self.assertFalse(got["block"])
+        self.assertFalse(got["configured"])
+        self.assertEqual(got["canonical_repo_root"], root)
+
+
+# ---------------------------------------------------------------------------
+# assess_canonical_repository_root: configured cross-repo path
+# ---------------------------------------------------------------------------
+class TestConfiguredCrossRepo(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = Path(self._tmp.name)
+        self.install = _init_repo(
+            self.tmp / "Gitea-Tools",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
+        )
+        self.target = _init_repo(
+            self.tmp / "mcp-control-plane",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
+        )
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_valid_configured_root_binds_to_target(self):
+        got = crr.assess_canonical_repository_root(
+            configured_value=self.target,
+            source="profile canonical_repository_root",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["proven"], got.get("reasons"))
+        self.assertFalse(got["block"])
+        self.assertTrue(got["configured"])
+        self.assertEqual(got["canonical_repo_root"], self.target)
+        self.assertNotEqual(got["canonical_repo_root"], self.install)
+
+    def test_missing_path_fails_closed(self):
+        got = crr.assess_canonical_repository_root(
+            configured_value=str(self.tmp / "does-not-exist"),
+            source="profile canonical_repository_root",
+            expected_slug=None,
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["block"])
+        self.assertFalse(got["proven"])
+        self.assertTrue(any("exist" in r for r in got["reasons"]))
+
+    def test_non_git_path_fails_closed(self):
+        plain = self.tmp / "plain"
+        plain.mkdir()
+        got = crr.assess_canonical_repository_root(
+            configured_value=str(plain),
+            source="profile canonical_repository_root",
+            expected_slug=None,
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["block"])
+        self.assertTrue(any("git" in r for r in got["reasons"]))
+
+    def test_identity_mismatch_fails_closed(self):
+        # Configured root is the install repo, but session expects the target
+        # repo identity: a forged/conflicting binding.
+        got = crr.assess_canonical_repository_root(
+            configured_value=self.install,
+            source="profile canonical_repository_root",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["block"])
+        self.assertTrue(any("identity" in r for r in got["reasons"]))
+
+    def test_unprovable_identity_blocks_when_required(self):
+        noremote = _init_repo(self.tmp / "noremote-src", "x", remote_name="origin")
+        _git(noremote, "remote", "remove", "origin")
+        got = crr.assess_canonical_repository_root(
+            configured_value=noremote,
+            source="profile canonical_repository_root",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.install,
+            require_binding=True,
+        )
+        self.assertTrue(got["block"])
+
+    def test_repository_identity_slug_reads_remote(self):
+        self.assertEqual(
+            crr.repository_identity_slug(self.target),
+            "Scaled-Tech-Consulting/mcp-control-plane",
+        )
+
+
+# ---------------------------------------------------------------------------
+# Workspace membership + branches-only enforced inside the TARGET repo (AC4)
+# ---------------------------------------------------------------------------
+class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = Path(self._tmp.name)
+        self.install = _init_repo(
+            self.tmp / "Gitea-Tools",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
+        )
+        self.target = _init_repo(
+            self.tmp / "mcp-control-plane",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
+        )
+        (Path(self.target) / "branches").mkdir()
+        self.target_wt = _add_worktree(
+            self.target,
+            Path(self.target) / "branches" / "author-issue-1",
+            "feat/issue-1",
+        )
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_context_canonical_root_is_configured_target(self):
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="author",
+            worktree_path=self.target_wt,
+            process_project_root=self.install,
+            env={},
+            configured_canonical_root=self.target,
+        )
+        self.assertEqual(ctx["canonical_repo_root"], self.target)
+        self.assertFalse(ctx["roots_aligned"])
+
+    def test_target_worktree_is_member_of_target_root(self):
+        got = nwb.amw.assess_workspace_repo_membership(
+            workspace_path=self.target_wt,
+            canonical_repo_root=self.target,
+        )
+        self.assertTrue(got["proven"], got.get("reasons"))
+
+    def test_target_worktree_not_member_of_install_root(self):
+        got = nwb.amw.assess_workspace_repo_membership(
+            workspace_path=self.target_wt,
+            canonical_repo_root=self.install,
+        )
+        self.assertTrue(got["block"])
+
+    def test_branches_guard_passes_inside_target(self):
+        assessment = nwb.assess_namespace_mutation_workspace(
+            role_kind="author",
+            worktree_path=self.target_wt,
+            worktree=None,
+            process_project_root=self.install,
+            env={},
+            current_branch="feat/issue-1",
+            configured_canonical_root=self.target,
+        )
+        self.assertFalse(assessment["block"], assessment.get("reasons"))
+        self.assertEqual(assessment["canonical_repo_root"], self.target)
+
+    def test_target_control_checkout_blocks_branches_guard(self):
+        assessment = nwb.assess_namespace_mutation_workspace(
+            role_kind="author",
+            worktree_path=self.target,  # stable target checkout, not branches/
+            worktree=None,
+            process_project_root=self.install,
+            env={},
+            current_branch="master",
+            configured_canonical_root=self.target,
+        )
+        self.assertTrue(assessment["block"])
+
+
+# ---------------------------------------------------------------------------
+# Immutable session pin (AC1/AC5)
+# ---------------------------------------------------------------------------
+class TestSessionCanonicalRootPin(unittest.TestCase):
+    def setUp(self):
+        os.environ["PYTEST_CURRENT_TEST"] = "t"
+        session_ctx._reset_session_context_for_testing()
+
+    def tearDown(self):
+        session_ctx._reset_session_context_for_testing()
+
+    def test_seed_stores_canonical_root(self):
+        ctx = session_ctx.seed_session_context_if_unbound(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            host="gitea.prgs.cc",
+            identity="svc",
+            repository="mcp-control-plane",
+            org="Scaled-Tech-Consulting",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        self.assertEqual(ctx["canonical_repository_root"], "/repo/mcp-control-plane")
+
+    def test_canonical_root_drift_fails_closed(self):
+        session_ctx.seed_session_context_if_unbound(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            host="gitea.prgs.cc",
+            identity="svc",
+            repository="mcp-control-plane",
+            org="Scaled-Tech-Consulting",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        got = session_ctx.assess_session_context(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            canonical_repository_root="/repo/forged-elsewhere",
+        )
+        self.assertTrue(got["block"])
+        self.assertTrue(any("canonical" in r for r in got["reasons"]))
+
+    def test_matching_canonical_root_passes(self):
+        session_ctx.seed_session_context_if_unbound(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            host="gitea.prgs.cc",
+            identity="svc",
+            repository="mcp-control-plane",
+            org="Scaled-Tech-Consulting",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        got = session_ctx.assess_session_context(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        self.assertFalse(got["block"], got["reasons"])
+
+
+# ---------------------------------------------------------------------------
+# Simultaneous prgs / mdcps isolation (AC6)
+# ---------------------------------------------------------------------------
+class TestSimultaneousIsolation(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = Path(self._tmp.name)
+        self.prgs = _init_repo(
+            self.tmp / "prgs-repo",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
+        )
+        self.mdcps = _init_repo(
+            self.tmp / "mdcps-repo",
+            "https://gitea.dadeschools.net/dadeschools/eAgenda.git",
+        )
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_each_namespace_binds_its_own_target(self):
+        a = crr.assess_canonical_repository_root(
+            configured_value=self.prgs,
+            source="env",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.tmp.as_posix(),
+        )
+        b = crr.assess_canonical_repository_root(
+            configured_value=self.mdcps,
+            source="env",
+            expected_slug="dadeschools/eAgenda",
+            process_project_root=self.tmp.as_posix(),
+        )
+        self.assertEqual(a["canonical_repo_root"], self.prgs)
+        self.assertEqual(b["canonical_repo_root"], self.mdcps)
+        self.assertNotEqual(a["canonical_repo_root"], b["canonical_repo_root"])
+
+    def test_cross_wired_identity_blocks(self):
+        # prgs path claimed under the mdcps identity → forged binding.
+        got = crr.assess_canonical_repository_root(
+            configured_value=self.prgs,
+            source="env",
+            expected_slug="dadeschools/eAgenda",
+            process_project_root=self.tmp.as_posix(),
+        )
+        self.assertTrue(got["block"])
+
+
+# ---------------------------------------------------------------------------
+# Config validation of the profile field (AC5: malformed bindings fail closed)
+# ---------------------------------------------------------------------------
+class TestConfigValidation(unittest.TestCase):
+    def test_absent_is_allowed(self):
+        # single-repo default: no field configured
+        gitea_config._validate_canonical_repository_root("p", None)
+
+    def test_valid_absolute_path_ok(self):
+        gitea_config._validate_canonical_repository_root(
+            "p", "/Users/x/Development/mcp-control-plane"
+        )
+
+    def test_relative_path_rejected(self):
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config._validate_canonical_repository_root(
+                "p", "relative/path"
+            )
+
+    def test_empty_string_rejected(self):
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config._validate_canonical_repository_root("p", "   ")
+
+    def test_non_string_rejected(self):
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config._validate_canonical_repository_root("p", ["/x"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_706_f1_seed_identity_integration.py

@@ -0,0 +1,221 @@
+"""Issue #706 F1 integration regression (review 457).
+
+The unit tests in ``test_issue_706_canonical_repository_root.py`` exercise
+``crr.assess_canonical_repository_root`` / ``session_ctx.seed_session_context_if_unbound``
+with a *preselected* slug, so they never drive the real end-to-end path that
+review 457 found broken:
+
+    _seed_session_context -> _trusted_session_repository -> _workspace_repository_slug
+    -> _local_git_remote_url (cwd=PROJECT_ROOT, always Gitea-Tools)
+
+Before the fix, the session repository identity was pinned from the *install*
+checkout remote even when a cross-repository ``canonical_repository_root`` was
+configured to an external repository (e.g. mcp-control-plane). The mutation
+preflight then derived ``expected_slug`` from that install-derived pin and
+``_enforce_canonical_repository_root`` failed closed on a self-inflicted
+identity mismatch, so the stated cross-repo namespaces stayed blocked.
+
+These tests use two *real* temporary git repositories and drive the live
+``mcp_server._seed_session_context`` and ``mcp_server._enforce_canonical_repository_root``
+functions, asserting the session pins the *configured target* identity and that
+enforcement accepts the target worktree — while every fail-closed property is
+preserved.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import mcp_server  # noqa: E402  # loads gitea_mcp_server.py into this namespace
+import canonical_repository_root as crr  # noqa: E402
+import session_context_binding as session_ctx  # noqa: E402
+
+INSTALL_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+TARGET_SLUG = "Scaled-Tech-Consulting/mcp-control-plane"
+TARGET_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git"
+OTHER_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/other-repo.git"
+
+
+def _git(cwd: str, *args: str) -> str:
+    return subprocess.run(
+        ["git", "-C", cwd, *args], capture_output=True, text=True, check=True
+    ).stdout.strip()
+
+
+def _init_repo(path: Path, remote_url: str) -> str:
+    path.mkdir(parents=True, exist_ok=True)
+    _git(str(path), "init", "-q")
+    _git(str(path), "config", "user.email", "[email protected]")
+    _git(str(path), "config", "user.name", "Test")
+    _git(str(path), "remote", "add", "prgs", remote_url)
+    (path / "README.md").write_text("seed\n")
+    _git(str(path), "add", "README.md")
+    _git(str(path), "commit", "-q", "-m", "seed")
+    return os.path.realpath(str(path))
+
+
+def _add_worktree(repo_root: str, wt: Path, branch: str) -> str:
+    _git(repo_root, "worktree", "add", "-q", "-b", branch, str(wt))
+    return os.path.realpath(str(wt))
+
+
+def _profile(role: str, *, canonical: str | None = None,
+             allowed=(TARGET_SLUG,)) -> dict:
+    p = {
+        "profile_name": f"mcp-control-plane-{role}",
+        "role": role,
+        "username": "svc",
+        "allowed_operations": ["gitea.read"],
+        "forbidden_operations": [],
+        "allowed_repositories": list(allowed),
+    }
+    if canonical is not None:
+        p["canonical_repository_root"] = canonical
+    return p
+
+
+class _Base(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        tmp = Path(self._tmp.name)
+        self.install = _init_repo(tmp / "Gitea-Tools", INSTALL_URL)
+        self.target = _init_repo(tmp / "mcp-control-plane", TARGET_URL)
+        (Path(self.target) / "branches").mkdir()
+        self.target_wt = _add_worktree(
+            self.target, Path(self.target) / "branches" / "author-issue-1",
+            "feat/issue-1",
+        )
+        # PROJECT_ROOT and the install git remote are the Gitea-Tools install
+        # checkout — exactly the source that (mis)seeded the session before.
+        self._p_root = mock.patch.object(mcp_server, "PROJECT_ROOT", self.install)
+        self._p_url = mock.patch.object(
+            mcp_server, "_local_git_remote_url", return_value=INSTALL_URL
+        )
+        self._p_root.start()
+        self._p_url.start()
+        session_ctx._reset_session_context_for_testing()
+
+    def tearDown(self):
+        mock.patch.stopall()
+        session_ctx._reset_session_context_for_testing()
+
+    def _seed(self, profile, env=None):
+        session_ctx._reset_session_context_for_testing()
+        with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
+                mock.patch.dict(os.environ, env or {}, clear=False):
+            return mcp_server._seed_session_context(
+                profile=profile, remote="prgs", host="gitea.prgs.cc",
+                identity="svc",
+            )
+
+    def _enforce(self, profile, env=None):
+        with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
+                mock.patch.dict(os.environ, env or {}, clear=False):
+            mcp_server._enforce_canonical_repository_root(
+                self.target_wt, remote="prgs"
+            )
+
+
+class TestF1SeedsConfiguredTargetIdentity(_Base):
+    def test_env_config_seeds_target_not_install(self):
+        ctx = self._seed(
+            _profile("reviewer"),
+            env={crr.CANONICAL_ROOT_ENV: self.target},
+        )
+        self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
+        self.assertEqual(ctx["repository"], "mcp-control-plane")
+        # Regression guard: must NOT be the install repo.
+        self.assertNotEqual(ctx["repository"], "Gitea-Tools")
+
+    def test_profile_field_seeds_target_not_install(self):
+        ctx = self._seed(_profile("merger", canonical=self.target))
+        self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
+        self.assertEqual(ctx["repository"], "mcp-control-plane")
+        self.assertNotEqual(ctx["repository"], "Gitea-Tools")
+
+    def test_enforce_accepts_target_after_seed_reviewer(self):
+        prof = _profile("reviewer", canonical=self.target)
+        self._seed(prof)
+        # Would raise RuntimeError on the self-inflicted identity mismatch
+        # before the fix.
+        self._enforce(prof)
+
+    def test_enforce_accepts_target_after_seed_merger(self):
+        prof = _profile("merger", canonical=self.target)
+        self._seed(prof)
+        self._enforce(prof)
+
+    def test_env_overrides_profile_for_seed(self):
+        # profile points at install; env points at the real target → env wins.
+        prof = _profile("reviewer", canonical=self.install,
+                        allowed=(TARGET_SLUG,))
+        ctx = self._seed(prof, env={crr.CANONICAL_ROOT_ENV: self.target})
+        self.assertEqual(ctx["repository"], "mcp-control-plane")
+
+
+class TestUnconfiguredDefaultUnchanged(_Base):
+    def test_unconfigured_keeps_install_identity(self):
+        prof = _profile("author", allowed=("Scaled-Tech-Consulting/Gitea-Tools",))
+        ctx = self._seed(prof)  # no env, no profile canonical field
+        self.assertEqual(ctx["repository"], "Gitea-Tools")
+        self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
+
+
+class TestFailClosed(_Base):
+    def _trusted(self, profile, env=None, *, for_mutation=True):
+        with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
+                mock.patch.dict(os.environ, env or {}, clear=False):
+            return mcp_server._trusted_session_repository(
+                profile, "prgs", for_mutation=for_mutation
+            )
+
+    def test_nonexistent_configured_root_fails_closed(self):
+        res = self._trusted(
+            _profile("reviewer"),
+            env={crr.CANONICAL_ROOT_ENV: self.target + "-missing"},
+        )
+        self.assertIsNone(res["repository"])
+        self.assertTrue(res["reasons"])
+
+    def test_non_git_configured_root_fails_closed(self):
+        plain = Path(self._tmp.name) / "plain"
+        plain.mkdir()
+        res = self._trusted(
+            _profile("reviewer"), env={crr.CANONICAL_ROOT_ENV: str(plain)}
+        )
+        self.assertIsNone(res["repository"])
+        self.assertTrue(any("git" in r for r in res["reasons"]))
+
+    def test_unallowlisted_target_identity_fails_closed(self):
+        # Configured target is valid, but the profile does not authorize it.
+        res = self._trusted(
+            _profile("reviewer", canonical=self.target,
+                     allowed=("Scaled-Tech-Consulting/Gitea-Tools",)),
+        )
+        self.assertIsNone(res["repository"])
+        self.assertTrue(any("scope" in r.lower() for r in res["reasons"]))
+
+    def test_forged_identity_conflict_fails_closed_at_enforce(self):
+        # Seed the target, then present a *different* configured root at
+        # enforcement time: the session pin no longer matches → fail closed.
+        other = _init_repo(Path(self._tmp.name) / "other", OTHER_URL)
+        prof_seed = _profile("reviewer", canonical=self.target)
+        self._seed(prof_seed)
+        prof_drift = _profile(
+            "reviewer", canonical=other,
+            allowed=(TARGET_SLUG, "Scaled-Tech-Consulting/other-repo"),
+        )
+        with self.assertRaises(RuntimeError):
+            self._enforce(prof_drift)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_preflight_workspace_repo_forwarding.py

@@ -0,0 +1,280 @@
+"""Regression matrix: session-bound mutation preflight resolves the workspace repo.
+
+Root-cause follow-up to #733. ``_resolve`` (the actual API-targeting resolver)
+already prefers the workspace-aligned git remote over the ``REMOTES`` default
+(bare ``prgs`` → ``Scaled-Tech-Consulting/Timesheet``). The shared #604
+anti-stomp preflight (``_run_anti_stomp_preflight``) did **not** mirror that: for
+every mutation task whose call site omitted ``org``/``repo`` it filled the
+remote-wide ``REMOTES`` default, so the #530 repo guard false-positived
+``wrong_repo`` against a ``Gitea-Tools`` checkout and blocked native
+issue/PR/review/merge/lock/cleanup mutations.
+
+The fix makes ``_run_anti_stomp_preflight`` prefer the trusted, git-remote-
+derived workspace identity for omitted coordinates on **session-bound** mutation
+tasks (marking those filled sides explicit, exactly like ``_resolve``), while
+``delete_branch`` keeps its explicit cross-repository *target* contract (#733)
+and continues to fail closed on omitted coordinates.
+"""
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import mcp_server
+import remote_repo_guard
+
+# Workspace is bound to Gitea-Tools even though the prgs REMOTES default repo is
+# Timesheet (the exact cross-repo scenario).
+LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+
+RECONCILER = {
+    "profile_name": "prgs-reconciler",
+    "role": "reconciler",
+    "allowed_operations": [
+        "gitea.read", "gitea.issue.comment", "gitea.pr.comment",
+        "gitea.pr.close", "gitea.branch.delete",
+    ],
+    "forbidden_operations": ["gitea.pr.create", "gitea.pr.merge"],
+    "audit_label": "prgs-reconciler",
+}
+
+# Session-bound mutation tasks (every MUTATION_TASK except the explicit-target
+# delete_branch) must auto-resolve the workspace repository for omitted coords.
+SESSION_BOUND_TASKS = [
+    "create_issue",
+    "comment_issue",
+    "close_issue",
+    "mark_issue",
+    "lock_issue",
+    "set_issue_labels",
+    "create_label",
+    "create_pr",
+    "close_pr",
+    "edit_pr",
+    "commit_files",
+    "cleanup_merged_pr_branch",
+    "cleanup_stale_claims",
+    "reconcile_merged_cleanups",
+    "reconcile_already_landed_pr",
+    "reconcile_close_superseded_pr",
+    "post_heartbeat",
+    "review_pr",
+    "submit_pr_review",
+    "merge_pr",
+]
+
+
+class _AntiStompResolutionBase(unittest.TestCase):
+    def setUp(self):
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.prgs.cc",
+                "org": "Scaled-Tech-Consulting",
+                "repo": "Timesheet",
+            },
+        })
+        self._remotes.start()
+
+    def tearDown(self):
+        patch.stopall()
+
+    def _capture_resolution(self, task, org, repo, *, local_url=LOCAL_GITEA_TOOLS_URL):
+        """Drive the live anti-stomp runner and capture the org/repo it resolved
+        and passed to the pure assessor."""
+        passthrough = {
+            "allowed": True, "block": False, "blockers": [], "reasons": [],
+            "exact_next_action": "proceed", "blocker_kind": None, "checks": {},
+        }
+        with patch.dict(
+            os.environ, {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, clear=False
+        ), patch(
+            "mcp_server._local_git_remote_url", return_value=local_url
+        ), patch(
+            "mcp_server.get_profile", return_value=RECONCILER
+        ), patch.object(
+            mcp_server.anti_stomp_preflight,
+            "assess_anti_stomp_preflight",
+            return_value=passthrough,
+        ) as m:
+            mcp_server._run_anti_stomp_preflight(
+                task, remote="prgs", org=org, repo=repo
+            )
+        self.assertTrue(m.called)
+        return m.call_args.kwargs
+
+
+class TestSessionBoundResolution(_AntiStompResolutionBase):
+    def test_omitted_coords_resolve_workspace_repo_and_mark_explicit(self):
+        for task in SESSION_BOUND_TASKS:
+            with self.subTest(task=task):
+                kw = self._capture_resolution(task, None, None)
+                self.assertEqual(
+                    kw["resolved_org"], "Scaled-Tech-Consulting", task
+                )
+                self.assertEqual(kw["resolved_repo"], "Gitea-Tools", task)
+                # Workspace-filled sides are intentional alignment → explicit.
+                self.assertTrue(kw["org_explicit"], task)
+                self.assertTrue(kw["repo_explicit"], task)
+
+    def test_resolved_workspace_repo_passes_the_530_guard(self):
+        kw = self._capture_resolution("create_pr", None, None)
+        assessment = remote_repo_guard.assess_remote_repo_match(
+            remote="prgs",
+            resolved_org=kw["resolved_org"],
+            resolved_repo=kw["resolved_repo"],
+            local_remote_url=LOCAL_GITEA_TOOLS_URL,
+            org_explicit=kw["org_explicit"],
+            repo_explicit=kw["repo_explicit"],
+        )
+        self.assertFalse(assessment["block"], assessment)
+
+    def test_explicit_caller_coords_are_preserved(self):
+        kw = self._capture_resolution(
+            "create_pr", "Scaled-Tech-Consulting", "Gitea-Tools"
+        )
+        self.assertEqual(kw["resolved_org"], "Scaled-Tech-Consulting")
+        self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
+        self.assertTrue(kw["org_explicit"])
+        self.assertTrue(kw["repo_explicit"])
+
+    def test_no_workspace_identity_falls_back_to_remotes_default(self):
+        # No derivable local remote → best-effort REMOTES default preserved,
+        # not marked explicit (so the guard still corroborates when able).
+        kw = self._capture_resolution("create_pr", None, None, local_url=None)
+        self.assertEqual(kw["resolved_repo"], "Timesheet")
+        self.assertFalse(kw["org_explicit"])
+        self.assertFalse(kw["repo_explicit"])
+
+
+class TestDeleteBranchContractPreserved(_AntiStompResolutionBase):
+    """delete_branch keeps its explicit-target contract (#733): omitted coords
+    still resolve the remote-wide default and are NOT marked explicit, so the
+    guard fails closed against a mismatched workspace."""
+
+    def test_omitted_coords_still_remote_default_not_explicit(self):
+        kw = self._capture_resolution("delete_branch", None, None)
+        self.assertEqual(kw["resolved_repo"], "Timesheet")
+        self.assertFalse(kw["org_explicit"])
+        self.assertFalse(kw["repo_explicit"])
+        assessment = remote_repo_guard.assess_remote_repo_match(
+            remote="prgs",
+            resolved_org=kw["resolved_org"],
+            resolved_repo=kw["resolved_repo"],
+            local_remote_url=LOCAL_GITEA_TOOLS_URL,
+            org_explicit=kw["org_explicit"],
+            repo_explicit=kw["repo_explicit"],
+        )
+        self.assertTrue(assessment["block"])
+
+    def test_explicit_coords_still_forwarded(self):
+        kw = self._capture_resolution(
+            "delete_branch", "Scaled-Tech-Consulting", "Gitea-Tools"
+        )
+        self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
+        self.assertTrue(kw["org_explicit"])
+        self.assertTrue(kw["repo_explicit"])
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class TestAuthorCallSiteForwardsOrgRepo(_AntiStompResolutionBase):
+    """#735: author mutation entrypoints must pass org/repo into preflight."""
+
+    def _assert_call_forwards(self, fn_name: str, *args, **kwargs):
+        import inspect
+        fn = getattr(mcp_server, fn_name)
+        # Force preflight path to raise with the kwargs it received so we can
+        # prove org/repo were forwarded without performing the full mutation.
+        captured = {}
+
+        def _capture(*a, **kw):
+            captured.update(kw)
+            captured["_args"] = a
+            raise RuntimeError("capture-only")
+
+        with patch.object(mcp_server, "verify_preflight_purity", side_effect=_capture), \
+             patch.object(mcp_server, "get_profile", return_value={
+                 "profile_name": "prgs-author",
+                 "role": "author",
+                 "allowed_operations": [
+                     "gitea.read", "gitea.issue.create", "gitea.issue.comment",
+                     "gitea.pr.create", "gitea.repo.commit", "gitea.branch.push",
+                 ],
+                 "forbidden_operations": [],
+                 "audit_label": "prgs-author",
+             }), patch.object(
+                 mcp_server, "_resolve",
+                 return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
+             ), patch.object(
+                 mcp_server, "_auth", return_value="token fake",
+             ), patch.object(
+                 mcp_server.role_session_router,
+                 "check_author_mutation_after_reviewer_stop",
+                 return_value=(True, []),
+             ), patch.object(
+                 mcp_server, "_namespace_mutation_block", return_value=None
+             ), patch.object(
+                 mcp_server, "_profile_permission_block", return_value=None
+             ), patch.object(
+                 mcp_server, "_production_guard_block_from_exc", return_value=None
+             ):
+            try:
+                fn(*args, **kwargs)
+            except RuntimeError as exc:
+                if "capture-only" not in str(exc):
+                    # Some tools re-raise; still require capture when preflight ran.
+                    if not captured:
+                        raise
+            self.assertTrue(captured, f"{fn_name} never called verify_preflight_purity")
+            self.assertEqual(captured.get("org"), "Scaled-Tech-Consulting", fn_name)
+            self.assertEqual(captured.get("repo"), "Gitea-Tools", fn_name)
+
+    def test_create_issue_forwards_explicit_org_repo(self):
+        self._assert_call_forwards(
+            "gitea_create_issue",
+            "title",
+            body="body with acceptance criteria for gate",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/fake-wt",
+        )
+
+    def test_create_pr_forwards_explicit_org_repo(self):
+        self._assert_call_forwards(
+            "gitea_create_pr",
+            title="t",
+            head="fix/issue-735-x",
+            base="master",
+            body="Closes #735",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/fake-wt",
+        )
+
+    def test_mark_issue_forwards_explicit_org_repo(self):
+        self._assert_call_forwards(
+            "gitea_mark_issue",
+            issue_number=735,
+            action="start",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/fake-wt",
+        )
+
+    def test_commit_files_forwards_explicit_org_repo(self):
+        self._assert_call_forwards(
+            "gitea_commit_files",
+            files=[{"operation": "update", "path": "x", "content_plain": "y"}],
+            message="m",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )