Execute Python code in a subprocess and capture output.
(
code: str,
timeout: int = 60,
task: Any | None = None,
on_progress: Any = None,
)
| 19 | from opc.layer4_tools.opaque_execution import ( |
| 20 | OpaqueExecutionEnvelopeError, |
| 21 | current_opaque_execution_plan, |
| 22 | verify_opaque_execution_plan, |
| 23 | ) |
| 24 | from opc.layer4_tools.registry import COMPANY_EFFECT_OPAQUE_EXACT, ToolDefinition |
| 25 | |
| 26 | |
| 27 | async def python_exec( |
| 28 | code: str, |
| 29 | timeout: int = 60, |
| 30 | task: Any | None = None, |
| 31 | on_progress: Any = None, |
| 32 | ) -> dict[str, Any]: |
| 33 | """Execute Python code in a subprocess and capture output.""" |
| 34 | _ = on_progress |
| 35 | frozen_plan = current_opaque_execution_plan("python_exec") |
| 36 | if frozen_plan is not None: |
| 37 | context = frozen_plan.context |
| 38 | roots = dict(frozen_plan.envelope.get("roots", {}) or {}) |
| 39 | workspace_root = str(roots.get("workspace_root", "") or "").strip() |
| 40 | if frozen_plan.preparation_error: |
| 41 | return { |
| 42 | "error": frozen_plan.preparation_error, |
| 43 | "exit_code": -1, |
| 44 | "execution_context": { |
| 45 | "workspace_root": workspace_root, |
| 46 | "python_executable": frozen_plan.executable, |
| 47 | "sandbox": frozen_plan.sandbox, |
| 48 | }, |
| 49 | } |
| 50 | timeout = frozen_plan.timeout |
| 51 | expected_code_digest = str( |
| 52 | frozen_plan.envelope.get("python_code_sha256", "") or "" |
| 53 | ) |
| 54 | actual_code_digest = hashlib.sha256(code.encode("utf-8")).hexdigest() |
| 55 | if expected_code_digest != actual_code_digest: |
| 56 | return { |
| 57 | "error": "company Python payload changed after approval", |
| 58 | "exit_code": -1, |
| 59 | "execution_context": { |
| 60 | "workspace_root": workspace_root, |
| 61 | "python_executable": frozen_plan.executable, |
| 62 | "sandbox": frozen_plan.sandbox, |
| 63 | }, |
| 64 | } |
| 65 | try: |
| 66 | verify_opaque_execution_plan(frozen_plan) |
| 67 | except OpaqueExecutionEnvelopeError as exc: |
| 68 | return { |
| 69 | "error": str(exc), |
| 70 | "exit_code": -1, |
| 71 | "execution_context": { |
| 72 | "workspace_root": workspace_root, |
| 73 | "python_executable": frozen_plan.executable, |
| 74 | "sandbox": frozen_plan.sandbox, |
| 75 | }, |
| 76 | } |
| 77 | else: |
| 78 | context = resolve_task_execution_context(task) |