from __future__ import annotations import hashlib import json import re from threading import Lock from typing import Any from sqlmodel import Session from ..core.redaction import redact_message, redact_value from ..core.time import utc_now from ..db.models import AppSetting from .action_execution import ( execute_approved_action, execute_governed_action, mcp_execution_supported, ) from .action_registry import ACTION_RISK, ACTIONS, action_capability from .approval_service import ( approval_fingerprint, get_approval, mark_approval_executed, ) from .audit_logger import log_event from .job_runner import cancel_job, get_job from .policy_engine import evaluate_policy _IDEMPOTENCY_LOCK = Lock() _IDEMPOTENCY_PREFIX = "mcp:operation:idempotency:" _IDEMPOTENCY_KEY = re.compile(r"^[A-Za-z0-9._:-]{8,128}$") def _reconciliation_required_response( *, action: str, target: str, risk: str, original_policy_decision: str | None, ) -> dict[str, Any]: return { "action": action, "target": target, "risk": risk, "policyDecision": "blocked", "originalPolicyDecision": original_policy_decision, "created": False, "executed": False, "executionSupported": mcp_execution_supported(action), "executionState": "reconciliation_required", "reason": ( "A previous MCP attempt crossed the durable execution boundary but did not " "persist a terminal response. Re-execution is refused to prevent a duplicate mutation." ), "nextStep": ( "Inspect the target, audit trail and any job state to reconcile the prior attempt; " "do not retry this idempotency key blindly." ), } def mcp_policy_source(action: str, target: str) -> str: """Normalize only the fixed acceptance authority at the authenticated MCP boundary.""" if (action in {"runner.acceptance.preflight", "runner.acceptance.quiesce"} and target == "tower/ci-acceptance-residue"): return "chatgpt-mcp" return "automation" def request_operation( session: Session, *, action: str, target: str, reason: str, idempotency_key: str, ) -> dict[str, Any]: """Execute an MCP mutation immediately when central policy permits it. MCP is intentionally a no-human-approval surface. A policy decision of ``approval_required`` is accepted only when AppOps has a typed executor for the exact action; it is then executed directly through that executor and audited with the original decision preserved as metadata. ``blocked`` actions stay blocked. Actions without a typed executor fail closed instead of creating an Approval row. A durable idempotency reservation is committed before any side effect. Once the reservation enters ``executing``, a replay without a terminal response is refused as ``reconciliation_required`` rather than risking a second mutation. """ safe_action = action.strip()[:160] safe_target = target.strip()[:240] safe_reason = redact_message(reason.strip())[:1000] if safe_action not in ACTIONS: raise ValueError("unknown_action") if not safe_target: raise ValueError("target_required") if not safe_reason: raise ValueError("reason_required") if not _IDEMPOTENCY_KEY.fullmatch(idempotency_key): raise ValueError("invalid_idempotency_key") capability = action_capability(safe_action) policy = evaluate_policy( session, action=safe_action, target=safe_target, source=mcp_policy_source(safe_action, safe_target), ) risk = str(policy.get("risk") or ACTION_RISK.get(safe_action, "critical")) if capability == "read_only": return { "action": safe_action, "target": safe_target, "risk": risk, "policyDecision": policy["decision"], "created": False, "executed": False, "executionSupported": False, "nextStep": "Use the corresponding read-only MCP tool.", } digest = hashlib.sha256( json.dumps( { "action": safe_action, "target": safe_target, "reason": safe_reason, }, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() row_key = ( f"{_IDEMPOTENCY_PREFIX}{hashlib.sha256(idempotency_key.encode()).hexdigest()}" ) with _IDEMPOTENCY_LOCK: row = session.get(AppSetting, row_key) if row: if row.value_json.get("requestDigest") != digest: raise ValueError("idempotency_key_reused_with_different_request") if row.value_json.get("response") is not None: return {**row.value_json.get("response", {}), "idempotentReplay": True} if row.value_json.get("state") == "executing": response = _reconciliation_required_response( action=safe_action, target=safe_target, risk=risk, original_policy_decision=row.value_json.get( "originalPolicyDecision" ), ) return {**response, "idempotentReplay": True} else: row = AppSetting( key=row_key, value_json={ "requestDigest": digest, "state": "reserved", "reservedAt": utc_now().isoformat(), "action": safe_action, "target": safe_target, }, is_secret=False, ) session.add(row) session.commit() if policy["decision"] == "blocked": audit = log_event( session, source="automation", identity="chatgpt-mcp", action_id=safe_action, target=safe_target, risk=risk, policy_decision="blocked", reason=policy["reason"], result="blocked", redacted_metadata={"integration": "mcp", "request": "operation"}, ) response = { "action": safe_action, "target": safe_target, "risk": risk, "policyDecision": "blocked", "reason": policy["reason"], "created": False, "executed": False, "executionSupported": False, "auditId": audit.id, "nextStep": "Blocked by AppOps policy. MCP does not create approvals.", } elif not mcp_execution_supported(safe_action): reason_text = ( "MCP no-human-approval mode has no typed executor for this action; " "the operation is blocked instead of creating an approval request." ) audit = log_event( session, source="automation", identity="chatgpt-mcp", action_id=safe_action, target=safe_target, risk=risk, policy_decision="blocked", reason=reason_text, result="blocked", redacted_metadata={ "integration": "mcp", "request": "operation", "originalPolicyDecision": policy["decision"], }, ) response = { "action": safe_action, "target": safe_target, "risk": risk, "policyDecision": "blocked", "originalPolicyDecision": policy["decision"], "reason": reason_text, "created": False, "executed": False, "executionSupported": False, "auditId": audit.id, "nextStep": "No MCP approval will be requested; add a typed executor if this action should become autonomous.", } else: original_decision = str(policy["decision"]) authorization_reason = safe_reason if original_decision == "approval_required": authorization_reason = redact_message( f"{safe_reason} MCP autonomous execution accepted the policy's approval-required posture without a human approval step: {policy['reason']}" )[:1000] # This commit is the side-effect boundary. If the process dies after this # point and before a terminal response is stored, a replay sees # state=executing and refuses to mutate again. row.value_json = { **row.value_json, "state": "executing", "executingAt": utc_now().isoformat(), "originalPolicyDecision": original_decision, } row.updated_at = utc_now() session.add(row) session.commit() result = execute_governed_action( session, action=safe_action, target=safe_target, source="chatgpt-mcp", risk=risk, reason=authorization_reason, original_policy_decision=original_decision, ) response = { **result, "action": safe_action, "target": safe_target, "risk": risk, "policyDecision": str(result.get("policyDecision") or "allowed"), "originalPolicyDecision": original_decision, "created": False, "executed": True, "executionSupported": True, "nextStep": ( "Use server.job.status for the returned job if further progress inspection is needed." if result.get("jobId") else "Operation completed through the typed MCP executor." ), } row.value_json = { **row.value_json, "state": "completed", "completedAt": utc_now().isoformat(), "response": response, } row.updated_at = utc_now() session.add(row) session.commit() return {**response, "idempotentReplay": False} def operation_status(session: Session, approval_id: str) -> dict[str, Any]: """Legacy status reader for approval records created before MCP no-approval mode.""" approval = get_approval(session, approval_id.strip()) if not approval: raise ValueError("approval_not_found") return redact_value( { "approvalId": approval.id, "action": approval.action, "target": approval.target, "risk": approval.risk, "status": approval.status, "actionFingerprint": approval.action_fingerprint, "expiresAt": approval.expires_at.isoformat() if approval.expires_at else None, "executionSupported": mcp_execution_supported(approval.action), "jobId": approval.job_id, "auditId": approval.audit_event_id, } ) def execute_operation( session: Session, *, approval_id: str, expected_action_fingerprint: str, ) -> dict[str, Any]: """Legacy executor for approval records created before MCP no-approval mode.""" approval = get_approval(session, approval_id.strip()) if not approval: raise ValueError("approval_not_found") if approval.status == "executed": return operation_status(session, approval.id) if approval.status != "approved": raise ValueError("approval_not_approved") if approval.expires_at is not None and approval.expires_at <= utc_now(): raise ValueError("approval_expired") expected = approval_fingerprint( approval.action, approval.target, approval.source, approval.risk, approval.request_context_json, ) supplied = expected_action_fingerprint.strip() if not supplied or supplied != expected or approval.action_fingerprint != expected: raise ValueError("approval_context_mismatch") if not mcp_execution_supported(approval.action): raise ValueError("operation_executor_unavailable") result = execute_approved_action(session, approval) if result.get("success"): mark_approval_executed( session, approval, job_id=str(result["jobId"]), audit_event_id=str(result["auditId"]), ) return redact_value({**result, "approvalId": approval.id}) def job_status(session: Session, job_id: str) -> dict[str, Any]: job = get_job(session, job_id.strip()) if not job: raise ValueError("job_not_found") return redact_value(job.model_dump(mode="json")) def cancel_operation_job(session: Session, job_id: str) -> dict[str, Any]: job = get_job(session, job_id.strip()) if not job: raise ValueError("job_not_found") from .privatemuse_deployment_recovery import ( privatemuse_recovery_job_cancellation_reason, ) cancellation_reason = privatemuse_recovery_job_cancellation_reason(session, job) if not cancellation_reason: from .workmesh_first_deployment import ( workmesh_first_deploy_job_cancellation_reason, ) cancellation_reason = workmesh_first_deploy_job_cancellation_reason(session, job) if cancellation_reason: return { "jobId": job.id, "status": job.status, "cancelled": False, "changed": False, "reason": cancellation_reason, } if job.status in {"succeeded", "failed", "cancelled"}: return { "jobId": job.id, "status": job.status, "cancelled": job.status == "cancelled", "changed": False, } previous = job.status updated = cancel_job(session, job, result_summary="Cancelled through governed MCP.") audit = log_event( session, source="automation", identity="chatgpt-mcp", action_id="job.cancel", target=updated.id, target_type="job", risk="low", policy_decision="allowed", reason="A governed MCP client cancelled a queued or running AppOps job.", result="success", before_state={"status": previous}, after_state={"status": updated.status}, redacted_metadata={"jobId": updated.id, "actionId": updated.action_id}, ) return { "jobId": updated.id, "status": updated.status, "cancelled": True, "changed": True, "auditId": audit.id, }