docs(opax): document staged MCP rollout workflow

- Add a private helper that performs an authenticated, read-only GET for the
  configured opax-mcp Cloud Run v2 service.
- Return only generation, observedGeneration, and updateTime on success, with
  sanitized reason codes for token, HTTP, network, timeout, and parsing errors.
- Add mocked unit coverage for successful responses and handled failure paths.

agents/mcp_server/README.md

@@ -56,6 +56,23 @@ gcloud builds submit \
   .
 ```
 
+### Observed Staged Rollout Behavior (2026-09-23)
+
+During the 2026-09-23 rollout, the `opax-mcp` service traffic configuration was
+explicitly pinned to a single production revision. Observations from that rollout
+include:
+
+- Candidate revisions created outside the tagged staged flow were observed to
+  retire before a later promotion decision.
+- A candidate deployed with both `--no-traffic` and a unique `--tag` remained
+  active and addressable for a health check.
+- Promotion required a separate, explicit `gcloud run services update-traffic`
+  command.
+
+These are observations from a specific rollout, not general Cloud Run
+guarantees. The staged workflow observed to succeed is documented in
+`docs/runbooks/opax-live-deploy.md`.
+
 ## CI1e — ADK-integrasjon (neste steg)
 
 Når `opax-mcp` er live, legges det til som ADK-tool i `agents/core-logic/agent.py`:

agents/mcp_server/server.py

@@ -18,6 +18,9 @@ CLOUD_RUN_SERVICE = os.environ.get("CLOUD_RUN_SERVICE", "osvauco-agent")
 MCP_SECRET        = os.environ.get("MCP_SECRET", "")
 BUILD_TRIGGER_ID  = os.environ.get("CLOUD_BUILD_TRIGGER_ID", "38423976-91ff-4ff4-859e-1f262344c609")
 
+_OPAX_MCP_SERVICE = "opax-mcp"
+_OPAX_MCP_REGION = "us-central1"
+
 
 def verify_token(request: Request, x_mcp_key: str = Header(default="")):
     # Aksepter både X-MCP-Key og api-key (Perplexity MCP connector bruker api-key)
@@ -70,6 +73,107 @@ def _parse_ts(ts):
         return None
 
 
+def _sanitize_opax_deployment_status(service_data: dict) -> dict:
+    """
+    Takes a decoded Cloud Run v2 service JSON object and returns a sanitized
+    dict conforming to the minimal v1 output contract.
+    """
+    def _validate_generation(val):
+        if isinstance(val, bool):
+            return None
+        if isinstance(val, int) and val >= 0:
+            return val
+        if isinstance(val, str) and val.isascii() and val.isdecimal():
+            return int(val)
+        return None
+
+    def _validate_update_time(val):
+        from datetime import datetime, timezone
+        if (
+            not isinstance(val, str)
+            or not val.endswith("Z")
+            or val.endswith("ZZ")
+            or len(val) < 20
+            or val[10] != "T"
+        ):
+            return None
+        try:
+            normalized_val = val[:-1] + "+00:00"
+            dt = datetime.fromisoformat(normalized_val)
+            if dt.tzinfo is None:
+                return None # Reject timezone-naive
+            return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+        except ValueError:
+            return None
+
+    return {
+        "service": _OPAX_MCP_SERVICE,
+        "region": _OPAX_MCP_REGION,
+        "generation": _validate_generation(service_data.get("generation")),
+        "observed_generation": _validate_generation(service_data.get("observedGeneration")),
+        "last_update_time": _validate_update_time(service_data.get("updateTime")),
+        "reason_code": None,
+    }
+
+
+def _fetch_opax_deployment_service_data() -> dict:
+    """
+    Fetches the raw Cloud Run v2 service object for OPAX-MCP, returning
+    only a minimal subset of fields or a sanitized failure reason.
+    """
+    import json
+    import socket
+    import urllib.error
+    import urllib.request
+
+    try:
+        token = _get_access_token()
+        if not token:
+            return {"reason_code": "ADC_TOKEN_UNAVAILABLE"}
+
+        endpoint_url = (
+            f"https://run.googleapis.com/v2/projects/{PROJECT_ID}/"
+            f"locations/{_OPAX_MCP_REGION}/services/{_OPAX_MCP_SERVICE}"
+        )
+        req = urllib.request.Request(
+            endpoint_url, headers={"Authorization": f"Bearer {token}"}, method="GET"
+        )
+
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            status = getattr(resp, "status", None)
+            if not isinstance(status, int):
+                return {"reason_code": "UNEXPECTED_LOCAL_FAILURE"}
+            if not (200 <= status < 300):
+                return {"reason_code": "UNEXPECTED_NON_2XX_STATUS"}
+
+            body_bytes = resp.read()
+
+        body = json.loads(body_bytes)
+        if not isinstance(body, dict):
+            return {"reason_code": "INVALID_RESPONSE_SHAPE"}
+
+        return {
+            "generation": body.get("generation"),
+            "observedGeneration": body.get("observedGeneration"),
+            "updateTime": body.get("updateTime"),
+        }
+
+    except json.JSONDecodeError:
+        return {"reason_code": "INVALID_JSON_RESPONSE"}
+    except urllib.error.HTTPError as e:
+        if e.code in (401, 403):
+            return {"reason_code": "CLOUD_RUN_UNAUTHORIZED"}
+        if e.code == 404:
+            return {"reason_code": "CLOUD_RUN_NOT_FOUND"}
+        return {"reason_code": "UNEXPECTED_NON_2XX_STATUS"}
+    except socket.timeout:
+        return {"reason_code": "REQUEST_TIMEOUT"}
+    except urllib.error.URLError:
+        return {"reason_code": "NETWORK_FAILURE"}
+    except Exception:
+        return {"reason_code": "UNEXPECTED_LOCAL_FAILURE"}
+
+
 class PushStaticRequest(BaseModel):
     file_path: str
     content: str

agents/mcp_server/test_server.py

@@ -0,0 +1,202 @@
+import unittest
+import json
+import socket
+import urllib.error
+from unittest.mock import patch, MagicMock
+from agents.mcp_server.server import (
+    _fetch_opax_deployment_service_data,
+    _sanitize_opax_deployment_status,
+)
+
+class TestSanitizeOpaxDeploymentStatus(unittest.TestCase):
+
+    def test_valid_inputs(self):
+        """Tests valid integer/string generations and a valid UTC timestamp."""
+        service_data = {
+            "generation": 10,
+            "observedGeneration": "9",
+            "updateTime": "2023-10-27T10:00:00Z"
+        }
+        result = _sanitize_opax_deployment_status(service_data)
+        self.assertEqual(result["generation"], 10)
+        self.assertEqual(result["observed_generation"], 9)
+        self.assertEqual(result["last_update_time"], "2023-10-27T10:00:00Z")
+        self.assertIsNone(result["reason_code"])
+
+    def test_zero_values(self):
+        """Tests handling of zero for generation values."""
+        service_data = {"generation": 0, "observedGeneration": "0"}
+        result = _sanitize_opax_deployment_status(service_data)
+        self.assertEqual(result["generation"], 0)
+        self.assertEqual(result["observed_generation"], 0)
+
+    def test_invalid_generation_values(self):
+        """Tests various invalid generation inputs."""
+        invalid_inputs = [
+            True, False, -1, 1.5, "-1", " 1 ", "abc", "١", None, {}, []
+        ]
+        for val in invalid_inputs:
+            with self.subTest(val=val):
+                res1 = _sanitize_opax_deployment_status({"generation": val})
+                res2 = _sanitize_opax_deployment_status({"observedGeneration": val})
+                self.assertIsNone(res1["generation"])
+                self.assertIsNone(res2["observed_generation"])
+
+    def test_fractional_second_timestamp_is_truncated(self):
+        """Tests that fractional seconds are correctly truncated."""
+        service_data = {"updateTime": "2023-10-27T10:00:00.123456Z"}
+        result = _sanitize_opax_deployment_status(service_data)
+        self.assertEqual(result["last_update_time"], "2023-10-27T10:00:00Z")
+
+    def test_invalid_and_non_z_timestamps(self):
+        """Tests invalid, tz-naive, missing, and non-Z timestamps."""
+        invalid_timestamps = [
+            "2023-10-27T10:00:00",          # Missing Z (naive)
+            "2023-10-27T10:00:00ZZ",         # Multiple Z
+            "2023-10-27T11:00:00+01:00",  # Valid but not Z-suffix
+            "2023-10-27 10:00:00Z",         # Malformed
+            "not-a-timestamp",              # Invalid
+            1698399600,                     # Not a string
+            None,                           # Missing
+        ]
+        for ts in invalid_timestamps:
+            with self.subTest(ts=ts):
+                result = _sanitize_opax_deployment_status({"updateTime": ts})
+                self.assertIsNone(result["last_update_time"])
+
+    def test_exact_key_set_and_fixed_values(self):
+        """Tests for exact key set and fixed service/region/reason_code."""
+        result = _sanitize_opax_deployment_status({})
+        expected_keys = {
+            "service", "region", "generation", "observed_generation",
+            "last_update_time", "reason_code"
+        }
+        self.assertEqual(set(result.keys()), expected_keys)
+        self.assertEqual(result["service"], "opax-mcp")
+        self.assertEqual(result["region"], "us-central1")
+        self.assertIsNone(result["reason_code"])
+
+
+class TestFetchOpaxDeploymentServiceData(unittest.TestCase):
+
+    @patch('urllib.request.urlopen')
+    @patch('agents.mcp_server.server._get_access_token')
+    def test_success_filters_fields(self, mock_get_token, mock_urlopen):
+        mock_get_token.return_value = "fake-token"
+        mock_response = MagicMock()
+        mock_response.status = 200
+        mock_response.read.return_value = json.dumps({
+            "generation": 5, "observedGeneration": 5, "updateTime": "T",
+            "uri": "forbidden"
+        }).encode()
+        mock_urlopen.return_value.__enter__.return_value = mock_response
+
+        result = _fetch_opax_deployment_service_data()
+        self.assertEqual(result, {"generation": 5, "observedGeneration": 5, "updateTime": "T"})
+        self.assertEqual(set(result.keys()), {"generation", "observedGeneration", "updateTime"})
+
+    @patch('urllib.request.urlopen')
+    @patch('agents.mcp_server.server._get_access_token')
+    def test_success_missing_fields(self, mock_get_token, mock_urlopen):
+        mock_get_token.return_value = "fake-token"
+        mock_response = MagicMock()
+        mock_response.status = 200
+        mock_response.read.return_value = json.dumps({"generation": 1}).encode()
+        mock_urlopen.return_value.__enter__.return_value = mock_response
+
+        result = _fetch_opax_deployment_service_data()
+        self.assertEqual(result, {"generation": 1, "observedGeneration": None, "updateTime": None})
+
+    @patch('urllib.request.urlopen')
+    @patch('agents.mcp_server.server._get_access_token')
+    def test_falsy_token_skips_call(self, mock_get_token, mock_urlopen):
+        for token in [None, ""]:
+            with self.subTest(token=token):
+                mock_get_token.return_value = token
+                result = _fetch_opax_deployment_service_data()
+                self.assertEqual(result, {"reason_code": "ADC_TOKEN_UNAVAILABLE"})
+                mock_urlopen.assert_not_called()
+
+    @patch('urllib.request.urlopen')
+    @patch('agents.mcp_server.server._get_access_token')
+    def test_get_token_exception(self, mock_get_token, mock_urlopen):
+        mock_get_token.side_effect = Exception("local failure")
+        result = _fetch_opax_deployment_service_data()
+        self.assertEqual(result, {"reason_code": "UNEXPECTED_LOCAL_FAILURE"})
+        mock_urlopen.assert_not_called()
+
+    @patch('urllib.request.urlopen')
+    @patch('agents.mcp_server.server._get_access_token')
+    def test_absent_response_status(self, mock_get_token, mock_urlopen):
+        mock_get_token.return_value = "fake-token"
+        class MockResp:
+            def __init__(self):
+                self.read_called = False
+            def read(self):
+                self.read_called = True
+                return b""
+            def __enter__(self):
+                return self
+            def __exit__(self, *args):
+                return False
+
+        mock_response = MockResp()
+        mock_urlopen.return_value = mock_response
+        result = _fetch_opax_deployment_service_data()
+        self.assertEqual(result, {"reason_code": "UNEXPECTED_LOCAL_FAILURE"})
+        self.assertFalse(mock_response.read_called)
+
+    @patch('urllib.request.urlopen')
+    @patch('agents.mcp_server.server._get_access_token')
+    def test_non_2xx_status_skips_read(self, mock_get_token, mock_urlopen):
+        mock_get_token.return_value = "fake-token"
+        mock_response = MagicMock()
+        mock_response.status = 503
+        mock_urlopen.return_value.__enter__.return_value = mock_response
+
+        result = _fetch_opax_deployment_service_data()
+        self.assertEqual(result, {"reason_code": "UNEXPECTED_NON_2XX_STATUS"})
+        mock_response.read.assert_not_called()
+
+    @patch('urllib.request.urlopen')
+    @patch('agents.mcp_server.server._get_access_token')
+    def test_http_and_network_errors(self, mock_get_token, mock_urlopen):
+        mock_get_token.return_value = "fake-token"
+        errors_to_reasons = [
+            (urllib.error.HTTPError(None, 401, "", {}, None), "CLOUD_RUN_UNAUTHORIZED"),
+            (urllib.error.HTTPError(None, 403, "", {}, None), "CLOUD_RUN_UNAUTHORIZED"),
+            (urllib.error.HTTPError(None, 404, "", {}, None), "CLOUD_RUN_NOT_FOUND"),
+            (urllib.error.HTTPError(None, 500, "", {}, None), "UNEXPECTED_NON_2XX_STATUS"),
+            (socket.timeout(), "REQUEST_TIMEOUT"),
+            (urllib.error.URLError("DNS failure"), "NETWORK_FAILURE"),
+        ]
+        for error, reason in errors_to_reasons:
+            with self.subTest(error=error.__class__.__name__, code=getattr(error, 'code', 'N/A')):
+                mock_urlopen.side_effect = error
+                result = _fetch_opax_deployment_service_data()
+                self.assertEqual(result, {"reason_code": reason})
+        mock_urlopen.side_effect = None
+
+    @patch('urllib.request.urlopen')
+    @patch('agents.mcp_server.server._get_access_token')
+    def test_invalid_json_body(self, mock_get_token, mock_urlopen):
+        mock_get_token.return_value = "fake-token"
+        mock_response = MagicMock()
+        mock_response.status = 200
+        mock_urlopen.return_value.__enter__.return_value = mock_response
+
+        # Malformed JSON
+        with self.subTest(case="malformed"):
+            mock_response.read.return_value = b'{"key":'
+            result = _fetch_opax_deployment_service_data()
+            self.assertEqual(result, {"reason_code": "INVALID_JSON_RESPONSE"})
+
+        # Valid JSON, but not a dictionary object
+        with self.subTest(case="non-object"):
+            mock_response.read.return_value = b'[1, 2, 3]'
+            result = _fetch_opax_deployment_service_data()
+            self.assertEqual(result, {"reason_code": "INVALID_RESPONSE_SHAPE"})
+
+
+if __name__ == '__main__':
+    unittest.main()

docs/runbooks/opax-live-deploy.md

@@ -34,6 +34,79 @@ to the existing Cloud Run service.
 
 ## MCP Deployment Behavior
 
+### OPAX-MCP Staged Deployment Workflow (Observed 2026-09-23)
+
+During the 2026-09-23 rollout, the service traffic configuration was explicitly
+pinned to a revision. Untagged zero-traffic candidate revisions were observed to
+retire before a later promotion decision. A candidate created with
+`--tag candidate-f003908 --no-traffic` remained active, addressable, was
+health-checked, and was explicitly promoted. These are observations from this
+rollout, not general Cloud Run guarantees.
+
+#### Stage A: Create and verify a tagged candidate
+
+1. Build and push an immutable image without deploying Cloud Run. The
+   `cloudbuild.build-only.yaml` configuration was used for this purpose in the
+   observed rollout.
+
+2. Create a tagged, no-traffic candidate revision:
+
+   ```bash
+   gcloud run deploy opax-mcp \
+     --image <immutable-image-digest> \
+     --region <region> \
+     --tag candidate-<short-commit> \
+     --no-traffic
+   ```
+
+3. Verify the candidate uses the expected image digest, is `Ready` and `Active`,
+   has a revision tag/direct URL, and has 0% traffic while the current
+   production revision retains 100%.
+
+4. Health-check only the tagged candidate URL:
+
+   ```bash
+   curl --silent --show-error --fail \
+     --max-time 10 \
+     --connect-timeout 5 \
+     --max-redirs 0 \
+     --request GET \
+     https://<candidate-tag-url>/health
+   ```
+
+   An empty result from a narrow Cloud Logging query means no data was returned
+   for that query; it does not prove the absence of application errors.
+
+#### Stage B: Promote traffic
+
+Promotion is a separate, explicit traffic action after Stage A verification:
+
+```bash
+gcloud run services update-traffic opax-mcp \
+  --to-revisions=<candidate-revision>=100 \
+  --region <region>
+```
+
+#### Rollback
+
+Rollback is also an explicit traffic action:
+
+```bash
+gcloud run services update-traffic opax-mcp \
+  --to-revisions=<last-known-good-revision>=100 \
+  --region <region>
+```
+
+#### Observed rollout record — 2026-09-23
+
+- **Source commit:** `f0039088c2505911c73e6a7374aa105415378277`
+- **Cloud Build ID:** `5b934520-8f8c-4a8e-89f3-f2f8377c2879`
+- **Immutable image:** `us-central1-docker.pkg.dev/propane-will-491900-m5/osvauco-repo/opax-mcp@sha256:1378a7b8cd81f7f82b52dafb29c0c9ee2b2d37800b27937fbdd353618f9319bf`
+- **Candidate tag/revision:** `candidate-f003908` / `opax-mcp-00218-duk`
+- **Rollback revision:** `opax-mcp-00214-nar`
+- **Health result:** HTTP 200 with `{"status":"ok","service":"opax-mcp","version":"3.6.0"}`
+- **Status:** Observed promoted successfully on 2026-09-23.
+
 ## Required OPAX-MCP Runtime Contract
 
 - `opax-mcp` uses Direct VPC egress to reach the internal Emma/Ollama runtime.

opax-mcp/gitea_handler.py

@@ -18,6 +18,10 @@ _BRANCH_RE = re.compile(
     r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$"
 )
 
+_READ_BRANCH_REF_RE = re.compile(
+    r"^[A-Za-z0-9][A-Za-z0-9._-]*(?:/[A-Za-z0-9][A-Za-z0-9._-]*)*$"
+)
+
 MAX_SOURCE_ARCHIVE_BYTES = int(
     os.environ.get("OPAX_MAX_SOURCE_ARCHIVE_BYTES", "104857600")
 )
@@ -188,6 +192,34 @@ def validate_path_for_write(path: str) -> str:
     return path
 
 
+def _validate_read_branch_ref(ref: str) -> str:
+    """Validate a safe branch name used only for read resolution."""
+    if not isinstance(ref, str) or not ref:
+        raise ValueError("Invalid git reference.")
+
+    if any(char.isspace() or ord(char) < 32 or ord(char) == 127 for char in ref):
+        raise ValueError("Invalid git reference.")
+
+    if (
+        ref.upper() == "HEAD"
+        or ref.startswith("refs/")
+        or ref.startswith("/")
+        or ref.endswith("/")
+        or "//" in ref
+        or "\\" in ref
+    ):
+        raise ValueError("Invalid git reference.")
+
+    components = ref.split("/")
+    if any(component in (".", "..") for component in components):
+        raise ValueError("Invalid git reference.")
+
+    if not _READ_BRANCH_REF_RE.fullmatch(ref):
+        raise ValueError("Invalid git reference.")
+
+    return ref
+
+
 async def resolve_branch_to_commit_sha(
     branch_name: str,
     repo_id: str,
@@ -342,10 +374,39 @@ async def handle_get_file_content(p: dict, server_repo_id: str) -> dict:
     if caller_repo is not None and caller_repo != validated_server_repo:
         raise ValueError("Repository file request is not allowed.")
 
-    ref = _validate_commit_sha(p.get("ref"))
     path = _validate_safe_path(p.get("path"))
 
-    url = f"{gitea_url}/api/v1/repos/{validated_server_repo}/contents/{quote(path, safe='')}?ref={ref}"
+    requested_ref = p.get("ref")
+    if not isinstance(requested_ref, str) or not requested_ref:
+        raise ValueError("Invalid git reference.")
+
+    if _SHA_RE.fullmatch(requested_ref.lower()):
+        resolved_commit_sha = _validate_commit_sha(requested_ref.lower())
+    else:
+        validated_branch_ref = _validate_read_branch_ref(requested_ref)
+        try:
+            resolved_commit_sha = await resolve_branch_to_commit_sha(
+                branch_name=validated_branch_ref,
+                repo_id=validated_server_repo,
+                gitea_url=gitea_url,
+            )
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code == 404:
+                raise ValueError(
+                    "Unknown or inaccessible branch reference."
+                ) from exc
+            logger.warning(
+                "Gitea branch resolution failed",
+                extra={"status_code": exc.response.status_code},
+            )
+            raise ValueError("Repository file is unavailable.") from exc
+        except Exception:
+            logger.error("Gitea branch resolution failed unexpectedly")
+            raise ValueError("Repository file is unavailable.")
+
+        resolved_commit_sha = _validate_commit_sha(resolved_commit_sha)
+
+    url = f"{gitea_url}/api/v1/repos/{validated_server_repo}/contents/{quote(path, safe='')}?ref={resolved_commit_sha}"
 
     try:
         async with httpx.AsyncClient(timeout=15) as c:
@@ -398,7 +459,14 @@ async def handle_get_file_content(p: dict, server_repo_id: str) -> dict:
     if '\0' in text_content:
         raise ValueError("Repository file content is not readable text.")
 
-    return {"path": path, "content": text_content, "encoding": "utf-8"}
+    return {
+        "repo": validated_server_repo,
+        "path": path,
+        "requested_ref": requested_ref,
+        "resolved_commit_sha": resolved_commit_sha,
+        "content": text_content,
+        "encoding": "utf-8",
+    }
 
 
 async def handle_list_repo_files(p: dict, default_repo: str) -> dict:

opax-mcp/server.py

@@ -100,13 +100,21 @@ class EncryptedPayloadV1(BaseModel):
     ciphertext: str
 
 
+GITEA_CHANGE_PLAN_STATUS_PENDING = "PENDING"
+GITEA_CHANGE_PLAN_STATUS_APPROVED = "APPROVED"
+GITEA_CHANGE_PLAN_STATUS_APPLYING = "APPLYING"
+GITEA_CHANGE_PLAN_STATUS_APPLIED = "APPLIED"
+GITEA_CHANGE_PLAN_STATUS_REJECTED = "REJECTED"
+GITEA_CHANGE_PLAN_STATUS_EXPIRED = "EXPIRED"
+
+
 class GiteaChangePlan(BaseModel):
     """Immutable admin-approved one-file Gitea change plan."""
 
     plan_id: str = Field(
         default_factory=lambda: f"gitea-change-{uuid.uuid4().hex}"
     )
-    status: str = "PENDING"
+    status: str = GITEA_CHANGE_PLAN_STATUS_PENDING
 
     created_at: datetime = Field(default_factory=datetime.utcnow)
     expires_at: datetime = Field(
@@ -134,6 +142,38 @@ class GiteaChangePlan(BaseModel):
     apply_error_message: Optional[str] = None
 
 
+def is_gitea_change_plan_expired(plan: GiteaChangePlan) -> bool:
+    """Checks if a Gitea change plan has expired."""
+    now_utc = datetime.now(timezone.utc)
+    expires_at = plan.expires_at
+
+    if expires_at.tzinfo is None:
+        expires_at = expires_at.replace(tzinfo=timezone.utc)
+
+    return now_utc >= expires_at
+
+
+_ALLOWED_GITEA_PLAN_TRANSITIONS = {
+    GITEA_CHANGE_PLAN_STATUS_PENDING: {
+        GITEA_CHANGE_PLAN_STATUS_APPROVED,
+        GITEA_CHANGE_PLAN_STATUS_REJECTED,
+        GITEA_CHANGE_PLAN_STATUS_EXPIRED,
+    },
+    GITEA_CHANGE_PLAN_STATUS_APPROVED: {
+        GITEA_CHANGE_PLAN_STATUS_APPLYING,
+        GITEA_CHANGE_PLAN_STATUS_EXPIRED,
+    },
+    GITEA_CHANGE_PLAN_STATUS_APPLYING: {GITEA_CHANGE_PLAN_STATUS_APPLIED},
+}
+
+def is_valid_gitea_change_plan_status_transition(
+    current_status: str,
+    next_status: str,
+) -> bool:
+    """Checks if a Gitea change plan status transition is allowed."""
+    return next_status in _ALLOWED_GITEA_PLAN_TRANSITIONS.get(current_status, set())
+
+
 async def create_gitea_change_plan(plan: GiteaChangePlan) -> dict:
     """Persist a pending one-file Gitea change plan in Firestore."""
     from google.cloud import firestore
@@ -230,6 +270,42 @@ async def get_gitea_change_plan(
     return GiteaChangePlan(**document.to_dict())
 
 
+async def transition_gitea_change_plan_status(
+    plan_id: str,
+    expected_status: str,
+    next_status: str,
+    updates: Optional[dict] = None,
+) -> GiteaChangePlan:
+    if not is_valid_gitea_change_plan_status_transition(expected_status, next_status):
+        raise ValueError("Invalid Gitea change plan status transition")
+    if updates and "status" in updates:
+        raise ValueError("Gitea change plan updates cannot include status")
+
+    from google.cloud import firestore
+
+    db = firestore.AsyncClient(project=GOOGLE_CLOUD_PROJECT)
+    plan_ref = db.collection("gitea_change_plans").document(plan_id)
+    transaction = db.transaction()
+
+    @firestore.async_transactional
+    async def transactional_update(transaction):
+        snapshot = await plan_ref.get(transaction=transaction)
+        if not snapshot.exists:
+            raise ValueError("Gitea change plan not found")
+        if snapshot.get("status") != expected_status:
+            raise ValueError("Gitea change plan status changed")
+
+        update_data = dict(updates or {})
+        update_data["status"] = next_status
+        transaction.update(plan_ref, update_data)
+
+    await transactional_update(transaction)
+    updated_plan = await get_gitea_change_plan(plan_id)
+    if updated_plan is None:
+        raise ValueError("Gitea change plan not found")
+    return updated_plan
+
+
 def _validate_admin_gitea_path(path: str) -> None:
     """Validate a repository-relative admin file path."""
     if not isinstance(path, str) or not path or path.isspace():
@@ -1744,4 +1820,4 @@ async def mcp_handler(request: Request):
 
 @app.get("/health")
 async def health():
-    return {"status": "ok", "service": "opax-mcp", "version": "3.6.0", "ollama": OLLAMA_BASE_URL} # BUMPED
+    return {"status": "ok", "service": "opax-mcp", "version": "3.6.0"}

opax-mcp/test_gitea_handler.py

@@ -3,12 +3,68 @@ from pathlib import Path
 import sys
 from unittest.mock import AsyncMock, MagicMock, patch
 import os
+import base64
+import json
 import httpx
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(REPO_ROOT / "opax-mcp"))
 
-from gitea_handler import handle_list_repo_files
+from gitea_handler import handle_list_repo_files, _validate_read_branch_ref, handle_get_file_content
+
+
+class TestValidateReadBranchRef(unittest.TestCase):
+    def test_accepted_refs(self):
+        accepted = [
+            "main",
+            "feat/opax-domain-decouple",
+            "docs/mcp-diagnostics",
+            "fix/example",
+            "chore/example",
+            "release/1.2.3",
+        ]
+        for ref in accepted:
+            with self.subTest(ref=ref):
+                self.assertEqual(_validate_read_branch_ref(ref), ref)
+
+    def test_rejected_refs(self):
+        rejected = [
+            "",
+            " main",
+            "main ",
+            "feature branch",
+            "main\tbranch",
+            "main\nbranch",
+            "main\x00branch",
+            "main\x7fbranch",
+            "HEAD",
+            "head",
+            "refs/heads/main",
+            "/main",
+            "main/",
+            "main//next",
+            "main/./next",
+            "main/../next",
+            "main\\next",
+            "main?x=1",
+            "main#fragment",
+            "https://example.invalid",
+            "user@host",
+            "feature*",
+            "feature~1",
+            "feature^",
+            "feature{a}",
+            "feature;cmd",
+            "feature|cmd",
+            "feature%2Fbranch",
+            "feature'quote",
+            'feature"quote',
+            "feature[abc]",
+        ]
+        for ref in rejected:
+            with self.subTest(ref=ref):
+                with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"):
+                    _validate_read_branch_ref(ref)
 
 class TestGiteaHandler(unittest.TestCase):
 
@@ -340,5 +396,332 @@ class TestGiteaHandler(unittest.TestCase):
         import asyncio
         asyncio.run(run_test())
 
+
+class TestHandleGetFileContent(unittest.TestCase):
+    def setUp(self):
+        self.original_gitea_url = os.environ.get("GITEA_URL")
+        self.original_gitea_token = os.environ.get("GITEA_TOKEN")
+        os.environ["GITEA_URL"] = "https://gitea.example.com"
+        os.environ["GITEA_TOKEN"] = "test-token"
+        self.server_repo_id = "chris/OSVauco"
+
+    def tearDown(self):
+        if self.original_gitea_url is None:
+            os.environ.pop("GITEA_URL", None)
+        else:
+            os.environ["GITEA_URL"] = self.original_gitea_url
+        if self.original_gitea_token is None:
+            os.environ.pop("GITEA_TOKEN", None)
+        else:
+            os.environ["GITEA_TOKEN"] = self.original_gitea_token
+
+    def _mock_contents_client(self, text):
+        body = json.dumps({
+            "content": base64.b64encode(text.encode("utf-8")).decode("ascii"),
+            "encoding": "base64",
+        }).encode("utf-8")
+
+        async def chunks():
+            yield body
+
+        response = MagicMock()
+        response.headers = {"content-length": str(len(body))}
+        response.aiter_bytes = chunks
+        stream = MagicMock()
+        stream.__aenter__.return_value = response
+        client = MagicMock()
+        client.stream.return_value = stream
+        async_client = MagicMock()
+        async_client.__aenter__.return_value = client
+        return async_client, client
+
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    @patch("httpx.AsyncClient")
+    def test_direct_lowercase_sha_success(
+        self, mock_async_client_constructor, mock_resolver
+    ):
+        import asyncio
+
+        sha = "a" * 40
+        mock_async_client, client = self._mock_contents_client("file content")
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            return await handle_get_file_content(
+                {
+                    "repo": self.server_repo_id,
+                    "path": "README.md",
+                    "ref": sha,
+                },
+                self.server_repo_id,
+            )
+
+        result = asyncio.run(run_test())
+
+        mock_resolver.assert_not_called()
+        client.stream.assert_called_once()
+        request_url = client.stream.call_args.args[1]
+        self.assertIn(f"ref={sha}", request_url)
+        self.assertEqual(result["requested_ref"], sha)
+        self.assertEqual(result["resolved_commit_sha"], sha)
+        self.assertEqual(result["content"], "file content")
+        self.assertEqual(result["encoding"], "utf-8")
+
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    @patch("httpx.AsyncClient")
+    def test_direct_uppercase_sha_normalizes_for_request(
+        self, mock_async_client_constructor, mock_resolver
+    ):
+        import asyncio
+
+        requested_sha = "A" * 40
+        resolved_sha = requested_sha.lower()
+        mock_async_client, client = self._mock_contents_client("file content")
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            return await handle_get_file_content(
+                {
+                    "repo": self.server_repo_id,
+                    "path": "README.md",
+                    "ref": requested_sha,
+                },
+                self.server_repo_id,
+            )
+
+        result = asyncio.run(run_test())
+
+        mock_resolver.assert_not_called()
+        request_url = client.stream.call_args.args[1]
+        self.assertIn(f"ref={resolved_sha}", request_url)
+        self.assertNotIn(f"ref={requested_sha}", request_url)
+        self.assertEqual(result["requested_ref"], requested_sha)
+        self.assertEqual(result["resolved_commit_sha"], resolved_sha)
+
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    @patch("httpx.AsyncClient")
+    def test_branch_resolves_to_sha_before_contents_request(
+        self, mock_async_client_constructor, mock_resolver
+    ):
+        import asyncio
+
+        resolved_sha = "b" * 40
+        mock_resolver.return_value = resolved_sha
+        mock_async_client, client = self._mock_contents_client("file content")
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            return await handle_get_file_content(
+                {
+                    "repo": self.server_repo_id,
+                    "path": "README.md",
+                    "ref": "main",
+                },
+                self.server_repo_id,
+            )
+
+        result = asyncio.run(run_test())
+
+        mock_resolver.assert_awaited_once_with(
+            branch_name="main",
+            repo_id=self.server_repo_id,
+            gitea_url="https://gitea.example.com",
+        )
+        request_url = client.stream.call_args.args[1]
+        self.assertIn(f"ref={resolved_sha}", request_url)
+        self.assertNotIn("ref=main", request_url)
+        self.assertEqual(result["repo"], self.server_repo_id)
+        self.assertEqual(result["path"], "README.md")
+        self.assertEqual(result["requested_ref"], "main")
+        self.assertEqual(result["resolved_commit_sha"], resolved_sha)
+        self.assertEqual(result["content"], "file content")
+        self.assertEqual(result["encoding"], "utf-8")
+
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    @patch("httpx.AsyncClient")
+    def test_unresolved_branch_raises_sanitized_error(
+        self, mock_async_client_constructor, mock_resolver
+    ):
+        import asyncio
+
+        mock_resolver.side_effect = httpx.HTTPStatusError(
+            "Not Found",
+            request=MagicMock(),
+            response=MagicMock(status_code=404),
+        )
+        mock_async_client, client = self._mock_contents_client("file content")
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            with self.assertRaisesRegex(
+                ValueError, r"^Unknown or inaccessible branch reference\.$"
+            ):
+                await handle_get_file_content(
+                    {
+                        "repo": self.server_repo_id,
+                        "path": "README.md",
+                        "ref": "missing-branch",
+                    },
+                    self.server_repo_id,
+                )
+
+        asyncio.run(run_test())
+
+        client.stream.assert_not_called()
+
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    @patch("httpx.AsyncClient")
+    def test_resolver_non_404_http_error_is_sanitized(
+        self, mock_async_client_constructor, mock_resolver
+    ):
+        import asyncio
+
+        mock_resolver.side_effect = httpx.HTTPStatusError(
+            "Server Error",
+            request=MagicMock(),
+            response=MagicMock(status_code=500),
+        )
+        mock_async_client, client = self._mock_contents_client("file content")
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            with self.assertRaisesRegex(
+                ValueError, r"^Repository file is unavailable\.$"
+            ):
+                await handle_get_file_content(
+                    {
+                        "repo": self.server_repo_id,
+                        "path": "README.md",
+                        "ref": "main",
+                    },
+                    self.server_repo_id,
+                )
+
+        asyncio.run(run_test())
+
+        client.stream.assert_not_called()
+
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    @patch("httpx.AsyncClient")
+    def test_resolver_unexpected_exception_is_sanitized(
+        self, mock_async_client_constructor, mock_resolver
+    ):
+        import asyncio
+
+        mock_resolver.side_effect = RuntimeError(
+            "upstream internal details must not reach the caller"
+        )
+        mock_async_client, client = self._mock_contents_client("file content")
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            with self.assertRaisesRegex(
+                ValueError, r"^Repository file is unavailable\.$"
+            ):
+                await handle_get_file_content(
+                    {
+                        "repo": self.server_repo_id,
+                        "path": "README.md",
+                        "ref": "main",
+                    },
+                    self.server_repo_id,
+                )
+
+        asyncio.run(run_test())
+
+        client.stream.assert_not_called()
+
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    @patch("httpx.AsyncClient")
+    def test_resolver_returns_malformed_sha_is_rejected(
+        self, mock_async_client_constructor, mock_resolver
+    ):
+        import asyncio
+
+        mock_resolver.return_value = "not-a-40-character-commit-sha"
+        mock_async_client, client = self._mock_contents_client("file content")
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            with self.assertRaisesRegex(ValueError, r"^Invalid commit SHA$"):
+                await handle_get_file_content(
+                    {
+                        "repo": self.server_repo_id,
+                        "path": "README.md",
+                        "ref": "main",
+                    },
+                    self.server_repo_id,
+                )
+
+        asyncio.run(run_test())
+
+        client.stream.assert_not_called()
+
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    @patch("httpx.AsyncClient")
+    def test_invalid_branch_ref_is_rejected(
+        self, mock_async_client_constructor, mock_resolver
+    ):
+        import asyncio
+
+        mock_async_client, client = self._mock_contents_client("file content")
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"):
+                await handle_get_file_content(
+                    {
+                        "repo": self.server_repo_id,
+                        "path": "README.md",
+                        "ref": "main/../secret",
+                    },
+                    self.server_repo_id,
+                )
+
+        asyncio.run(run_test())
+
+        mock_resolver.assert_not_called()
+        client.stream.assert_not_called()
+
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    @patch("httpx.AsyncClient")
+    def test_contents_api_404_is_sanitized(
+        self, mock_async_client_constructor, mock_resolver
+    ):
+        import asyncio
+
+        resolved_sha = "c" * 40
+        mock_resolver.return_value = resolved_sha
+        mock_async_client, client = self._mock_contents_client("file content")
+
+        response_mock = client.stream.return_value.__aenter__.return_value
+        response_mock.raise_for_status.side_effect = httpx.HTTPStatusError(
+            "Not Found", request=MagicMock(), response=MagicMock(status_code=404)
+        )
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            with self.assertRaisesRegex(
+                ValueError, r"^Repository file is unavailable\.$"
+            ):
+                await handle_get_file_content(
+                    {
+                        "repo": self.server_repo_id,
+                        "path": "README.md",
+                        "ref": "main",
+                    },
+                    self.server_repo_id,
+                )
+
+        asyncio.run(run_test())
+
+        mock_resolver.assert_awaited_once_with(
+            branch_name="main",
+            repo_id=self.server_repo_id,
+            gitea_url="https://gitea.example.com",
+        )
+        client.stream.assert_called_once()
+
+
 if __name__ == '__main__':
     unittest.main()

opax-mcp/test_gitea_helpers.py

@@ -297,11 +297,19 @@ class TestGetFileHardening(unittest.IsolatedAsyncioTestCase):
                     )
                 client.stream.assert_not_called()
 
-    async def test_get_file_ref_validation(self):
-        with self.assertRaisesRegex(ValueError, "Invalid commit SHA"):
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    async def test_get_file_ref_validation(self, mock_resolver):
+        mock_resolver.side_effect = httpx.ConnectError("Network error")
+        with self.assertRaisesRegex(ValueError, r"^Repository file is unavailable\.$"):
             await gitea_handler.handle_get_file_content({"path": "README.md", "ref": "main"}, "chris/OSVauco")
-        with self.assertRaisesRegex(ValueError, "Invalid commit SHA"):
+        mock_resolver.assert_awaited_once()
+        self.assertEqual(mock_resolver.call_args.kwargs['branch_name'], "main")
+        self.assertEqual(mock_resolver.call_args.kwargs['repo_id'], "chris/OSVauco")
+
+        with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"):
              await gitea_handler.handle_get_file_content({"path": "README.md", "ref": None}, "chris/OSVauco")
+        # Verify the resolver was not called a second time for the invalid ref
+        self.assertEqual(mock_resolver.await_count, 1)
 
     async def test_gitea_repo_validation(self):
         with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"):
@@ -309,17 +317,24 @@ class TestGetFileHardening(unittest.IsolatedAsyncioTestCase):
         with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"):
             await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "")
 
+    async def test_get_file_path_not_allowed(self):
+        for path in ["../secrets.txt", "/etc/passwd"]:
+            with self.subTest(path=path):
+                with self.assertRaisesRegex(ValueError, r"^PATH_NOT_ALLOWED$"):
+                    await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco")
+
+    async def test_get_file_secret_path_denied(self):
+        with self.assertRaisesRegex(ValueError, r"^SECRET_PATH_DENIED$"):
+            await gitea_handler.handle_get_file_content({"path": "docs/.env", "ref": self.valid_sha}, "chris/OSVauco")
+
     @patch("gitea_handler.httpx.AsyncClient")
-    async def test_get_file_path_validation(self, mock_client):
+    async def test_get_file_valid_paths(self, mock_client):
         client = mock_client.return_value.__aenter__.return_value
         configure_stream(client, json.dumps({'content': 'YQ==', 'size': 1}).encode("utf-8"))
-        for path in ["../secrets.txt", "/etc/passwd", "src/main.py", "docs/.env", "file.json"]:
-            with self.subTest(path=path):
-                with self.assertRaisesRegex(ValueError, "Repository file request is not allowed."):
-                    await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco")
         for path in ["README.md", "docs/ARCHITECTURE.md", ".gemini/GEMINI.md"]:
             with self.subTest(path=path):
-                 await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco")
+                await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco")
+        self.assertEqual(client.stream.call_count, 3)
 
     @patch("gitea_handler.httpx.AsyncClient")
     async def test_get_file_size_limit(self, mock_client):

opax-mcp/test_propose_gitea_change.py

@@ -18,6 +18,26 @@ KMS_KEY_NAME = "projects/p/locations/l/keyRings/k/cryptoKeys/k"
 
 class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
 
+    @patch('server.handle_get_file_content', new_callable=AsyncMock)
+    async def test_get_file_wrapper_enforces_server_repo(self, mock_handler):
+        sentinel_response = {
+            "requested_ref": "main",
+            "resolved_commit_sha": "a" * 40,
+            "content": "file content"
+        }
+        mock_handler.return_value = sentinel_response
+
+        params = {
+            "path": "README.md",
+            "ref": "main",
+            "repo": "untrusted/repo",
+        }
+        result = await server.get_file(params)
+
+        mock_handler.assert_awaited_once_with(params, server.GITEA_REPO)
+        self.assertIs(result, sentinel_response)
+
+
     def setUp(self):
         """Set up valid parameters for tests."""
         self.valid_params = {
@@ -344,5 +364,252 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
             self.assertNotEqual(base_hash, server._calculate_approval_subject_hash(modified_plan_cipher))
 
 
+class TestIsGiteaChangePlanExpired(unittest.TestCase):
+    def _create_test_plan(self, expires_at):
+        return server.GiteaChangePlan(
+            repo="o/r",
+            branch="b",
+            path="p",
+            base_sha="a" * 40,
+            content_hash="c" * 64,
+            commit_message="m",
+            unified_diff="d",
+            expires_at=expires_at,
+        )
+
+    def test_future_aware_is_not_expired(self):
+        from datetime import datetime, timedelta, timezone
+        future_time = datetime.now(timezone.utc) + timedelta(days=1)
+        plan = self._create_test_plan(future_time)
+        self.assertFalse(server.is_gitea_change_plan_expired(plan))
+
+    def test_past_aware_is_expired(self):
+        from datetime import datetime, timedelta, timezone
+        past_time = datetime.now(timezone.utc) - timedelta(days=1)
+        plan = self._create_test_plan(past_time)
+        self.assertTrue(server.is_gitea_change_plan_expired(plan))
+
+    def test_past_naive_is_expired(self):
+        from datetime import datetime, timedelta
+        past_time_naive = datetime.utcnow() - timedelta(days=1)
+        plan = self._create_test_plan(past_time_naive)
+        self.assertTrue(server.is_gitea_change_plan_expired(plan))
+
+    def test_default_status_is_pending_constant(self):
+        from datetime import datetime, timedelta, timezone
+        future_time = datetime.now(timezone.utc) + timedelta(days=1)
+        plan = self._create_test_plan(future_time)
+        self.assertEqual(
+            plan.status,
+            server.GITEA_CHANGE_PLAN_STATUS_PENDING,
+        )
+        self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_PENDING, "PENDING")
+        self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_APPROVED, "APPROVED")
+        self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_APPLYING, "APPLYING")
+        self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_APPLIED, "APPLIED")
+        self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_REJECTED, "REJECTED")
+        self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_EXPIRED, "EXPIRED")
+
+
+class TestGiteaChangePlanStatusTransitions(unittest.TestCase):
+    def test_allowed_transitions(self):
+        self.assertTrue(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_PENDING,
+                server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+            )
+        )
+        self.assertTrue(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_PENDING,
+                server.GITEA_CHANGE_PLAN_STATUS_REJECTED,
+            )
+        )
+        self.assertTrue(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_PENDING,
+                server.GITEA_CHANGE_PLAN_STATUS_EXPIRED,
+            )
+        )
+        self.assertTrue(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+                server.GITEA_CHANGE_PLAN_STATUS_APPLYING,
+            )
+        )
+        self.assertTrue(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+                server.GITEA_CHANGE_PLAN_STATUS_EXPIRED,
+            )
+        )
+        self.assertTrue(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_APPLYING,
+                server.GITEA_CHANGE_PLAN_STATUS_APPLIED,
+            )
+        )
+
+    def test_disallowed_transitions(self):
+        self.assertFalse(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_PENDING,
+                server.GITEA_CHANGE_PLAN_STATUS_PENDING,
+            )
+        )
+        self.assertFalse(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+                server.GITEA_CHANGE_PLAN_STATUS_APPLIED,
+            )
+        )
+        self.assertFalse(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_APPLYING,
+                server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+            )
+        )
+        self.assertFalse(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_APPLIED,
+                server.GITEA_CHANGE_PLAN_STATUS_APPLYING,
+            )
+        )
+        self.assertFalse(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_REJECTED,
+                server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+            )
+        )
+        self.assertFalse(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_EXPIRED,
+                server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+            )
+        )
+        self.assertFalse(
+            server.is_valid_gitea_change_plan_status_transition(
+                "UNKNOWN", server.GITEA_CHANGE_PLAN_STATUS_PENDING
+            )
+        )
+        self.assertFalse(
+            server.is_valid_gitea_change_plan_status_transition(
+                server.GITEA_CHANGE_PLAN_STATUS_PENDING, "UNKNOWN"
+            )
+        )
+
+
+class TestTransitionGiteaChangePlanStatus(unittest.IsolatedAsyncioTestCase):
+    @patch("google.cloud.firestore.AsyncClient")
+    async def test_rejects_invalid_transition_before_firestore(self, mock_db_client):
+        with self.assertRaisesRegex(
+            ValueError, r"^Invalid Gitea change plan status transition$"
+        ):
+            await server.transition_gitea_change_plan_status(
+                plan_id="plan-123",
+                expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING,
+                next_status=server.GITEA_CHANGE_PLAN_STATUS_APPLIED,
+            )
+        mock_db_client.assert_not_called()
+
+    @patch("google.cloud.firestore.AsyncClient")
+    async def test_rejects_status_in_updates_before_firestore(self, mock_db_client):
+        with self.assertRaisesRegex(
+            ValueError, r"^Gitea change plan updates cannot include status$"
+        ):
+            await server.transition_gitea_change_plan_status(
+                plan_id="plan-123",
+                expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING,
+                next_status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+                updates={"status": server.GITEA_CHANGE_PLAN_STATUS_REJECTED},
+            )
+        mock_db_client.assert_not_called()
+
+    @patch("server.get_gitea_change_plan", new_callable=AsyncMock)
+    @patch("google.cloud.firestore.async_transactional")
+    @patch("google.cloud.firestore.AsyncClient")
+    async def test_transitions_matching_status_and_returns_refetched_plan(
+        self, mock_db_client, mock_transactional, mock_get_plan
+    ):
+        def transactional_side_effect(callback):
+            async def wrapped(transaction):
+                return await callback(transaction)
+            return wrapped
+        mock_transactional.side_effect = transactional_side_effect
+
+        mock_transaction = MagicMock()
+        mock_transaction.update = MagicMock()
+        mock_db_client.return_value.transaction.return_value = mock_transaction
+
+        snapshot = MagicMock()
+        snapshot.exists = True
+        snapshot.get.return_value = server.GITEA_CHANGE_PLAN_STATUS_PENDING
+
+        mock_plan_ref = MagicMock()
+        mock_plan_ref.get = AsyncMock(return_value=snapshot)
+        mock_db_client.return_value.collection.return_value.document.return_value = mock_plan_ref
+
+        final_plan = server.GiteaChangePlan(
+            status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+            repo="o/r", branch="b", path="p", base_sha="a"*40,
+            content_hash="c"*64, commit_message="m", unified_diff="d",
+            approved_by="admin@example.com"
+        )
+        mock_get_plan.return_value = final_plan
+
+        updates = {"approved_by": "admin@example.com"}
+        result = await server.transition_gitea_change_plan_status(
+            plan_id="plan-123",
+            expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING,
+            next_status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+            updates=updates,
+        )
+
+        mock_transaction.update.assert_called_once_with(
+            mock_plan_ref,
+            {
+                "approved_by": "admin@example.com",
+                "status": server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+            },
+        )
+        self.assertEqual(updates, {"approved_by": "admin@example.com"})
+        mock_get_plan.assert_awaited_once_with("plan-123")
+        self.assertIs(result, final_plan)
+
+    @patch("server.get_gitea_change_plan", new_callable=AsyncMock)
+    @patch("google.cloud.firestore.async_transactional")
+    @patch("google.cloud.firestore.AsyncClient")
+    async def test_rejects_stale_status_inside_transaction(
+        self, mock_db_client, mock_transactional, mock_get_plan
+    ):
+        def transactional_side_effect(callback):
+            async def wrapped(transaction):
+                return await callback(transaction)
+            return wrapped
+        mock_transactional.side_effect = transactional_side_effect
+
+        mock_transaction = MagicMock()
+        mock_transaction.update = MagicMock()
+        mock_db_client.return_value.transaction.return_value = mock_transaction
+
+        snapshot = MagicMock()
+        snapshot.exists = True
+        snapshot.get.return_value = server.GITEA_CHANGE_PLAN_STATUS_APPROVED
+
+        mock_plan_ref = MagicMock()
+        mock_plan_ref.get = AsyncMock(return_value=snapshot)
+        mock_db_client.return_value.collection.return_value.document.return_value = mock_plan_ref
+
+        with self.assertRaisesRegex(ValueError, r"^Gitea change plan status changed$"):
+            await server.transition_gitea_change_plan_status(
+                plan_id="plan-123",
+                expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING,
+                next_status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED,
+            )
+
+        mock_transaction.update.assert_not_called()
+        mock_get_plan.assert_not_awaited()
+
+
 if __name__ == "__main__":
     unittest.main()