diff --git a/skills/iac-code/SKILL.md b/skills/iac-code/SKILL.md index 674c1530..6d503f37 100644 --- a/skills/iac-code/SKILL.md +++ b/skills/iac-code/SKILL.md @@ -29,6 +29,7 @@ Treat a bridge error returned before job creation as the authoritative outcome f ```json { "channel": "codex", + "pipelineName": "selling_solution_first", "permissionWaitPolicy": { "residentTimeoutSeconds": null, "subPipelineTimeoutSeconds": null, @@ -37,10 +38,10 @@ Treat a bridge error returned before job creation as the authoritative outcome f } ``` - `channel` stores only the channel identifier; the bridge adds the `skill/` prefix before sending it to iac-code. `permissionWaitPolicy` applies only to the temporary A2A server owned by this Skill: `null` timeouts mean unlimited waits, positive finite values set resident/Sub Pipeline limits, and grace is a non-negative finite value. Finite values cannot exceed 10 years; use `null` instead of an arbitrarily large number for an unlimited resident or Sub Pipeline wait. The bridge validates and converts this object into server configuration; it never sends the policy through A2A message metadata. Missing fields use the defaults shown above. The bridge rejects unknown configuration fields. If the file or a field is absent, no corresponding override is applied. Never derive these values from the user's request, ask the user for them, or create, edit, or reveal this install-local configuration during an infrastructure task. + `channel` stores only the channel identifier; the bridge adds the `skill/` prefix before sending it to iac-code. `pipelineName` selects the implementation used only after Pipeline mode is chosen: `selling_solution_first` is the default, while the legacy `selling` flow is used only when this install-local file explicitly selects it. `permissionWaitPolicy` applies only to the temporary A2A server owned by this Skill: `null` timeouts mean unlimited waits, positive finite values set resident/Sub Pipeline limits, and grace is a non-negative finite value. Finite values cannot exceed 10 years; use `null` instead of an arbitrarily large number for an unlimited resident or Sub Pipeline wait. The bridge validates and converts this object into server configuration; it never sends the policy through A2A message metadata. The bridge rejects unknown configuration fields and invalid Pipeline names. If `config.json` or `pipelineName` is absent, Pipeline mode uses `selling_solution_first`; other absent fields keep their existing defaults. Never derive these values from the user's request, ask the user for them, or create, edit, or reveal this install-local configuration during an infrastructure task. - Normal is the default, including concrete resource queries/changes, template work, troubleshooting, and deployment of a clear target. Use `--mode pipeline --pipeline-name selling` only when the user explicitly requests it or the request genuinely needs the fixed candidate-architecture, cost-comparison, plan-confirmation, and deployment flow. Questions, permissions, tool use, or deployment alone do not select Pipeline. When uncertain, use normal. - Start performs a non-secret configuration preflight through the Runtime. An incomplete LLM provider/API Key returns `llm_not_configured` and stops before creating a job. Selling Pipeline also requires complete Alibaba Cloud credentials and otherwise returns `cloud_credentials_not_configured`. Normal mode may continue without cloud credentials for work that does not call cloud APIs; report its preflight warning rather than claiming cloud operations are available. + Normal is the overall default, including concrete resource queries/changes, template work, troubleshooting, and deployment of a clear target. Use `--mode pipeline` only when the user explicitly requests it or the request genuinely needs the candidate-architecture, cost-comparison, plan-confirmation, and deployment flow. Pipeline mode uses solution-first unless the installed configuration explicitly selects legacy selling. Questions, permissions, tool use, or deployment alone do not select Pipeline. When uncertain, use normal. + Start performs a non-secret configuration preflight through the Runtime. An incomplete LLM provider/API Key returns `llm_not_configured` and stops before creating a job. Both supported Pipelines require complete Alibaba Cloud credentials and otherwise return `cloud_credentials_not_configured`. Normal mode may continue without cloud credentials for work that does not call cloud APIs; report its preflight warning rather than claiming cloud operations are available. 3. `--follow` consumes the event stream until the next parent/candidate step boundary, permission, user question, candidate selection, `turn_completed`, or terminal state. It writes every parent `step_started`/`step_completed`/`step_failed` and candidate `candidate_step_started`/`candidate_step_completed`/`candidate_step_failed` boundary plus low-frequency bounded heartbeats to stderr; stdout contains one bounded JSON result. A boundary result sets `boundaryReached: true`, `presentationRequired: true`, and provides ready-to-display localized strings in `userUpdates`. Before invoking another tool, emit every `userUpdates` string in a user-visible assistant text block, including the Step 1/2 conclusion already embedded in completed-step updates. Never leave these updates only in reasoning, Bash output, a tool description, or the final summary. After that visible text block, immediately call `follow` again with the returned cursor. Do not treat `boundaryReached` as completion. Do not expand this into raw tool-event or token-delta output. While it is running, do not independently answer the infrastructure task or ask a parallel business question. Only ask the user when the current result contains `inputRequired`. 4. If follow reaches its bounded wait window, call the diagnostic follow command again with the returned cursor: @@ -50,6 +51,7 @@ Treat a bridge error returned before job creation as the authoritative outcome f ``` The recommended wait is 60 seconds and the bridge enforces a 120-second maximum even if a larger value is supplied. + If a result says `state: input-required` but does not contain `inputRequired`, there is no user boundary to answer. Report its `latestText` or error, keep the same job unchanged, and stop. Never call `continue`, repeat `respond`, call `cancel`, or start a replacement job unless the user explicitly requests that action. 5. When `state` is `turn_completed`, treat `finalText` and `artifacts` as the authoritative normal-turn result. When a Pipeline reaches any terminal state, including `completed`, `failed`, `canceled`, or `rejected`, treat `pipelineResult` and `artifacts` as its authoritative result and present its success or failure details directly. If rollback cleanup is pending, the bridge automatically runs a cleanup-only normal task in the same context before returning the Pipeline result; keep following it and handle any returned permission normally. If cleanup is `failed` or `unavailable`, report that manual inspection or retry is required and do not claim it succeeded. Never send a synthetic cleanup prompt or a follow-up merely to retrieve or summarize an existing result. Never recover an answer from Session files, spool files, logs, or raw tool-result files. 6. To send the next natural-language message in the same normal conversation, or after a completed Pipeline has handed the same conversation to normal mode, write it to another workspace prompt file and continue the existing job: @@ -68,24 +70,20 @@ python3 scripts/iac_code.py poll --job-id --cursor --wait-seco ## User input -When `inputRequired` is present, preserve every correlation field in the response. Never reuse an answer file from another request. +Treat every `inputRequired` as a hard user-interaction boundary. Present it through the outer Agent's native user-question or approval UI and stop until the user explicitly answers that specific boundary. If no native UI is available, ask in a visible assistant turn and stop. Never infer, recommend-and-select, or submit an answer from the original infrastructure request, a prior answer, an outer tool-execution approval, a default, or the fact that only one option is available. Do not write an answer file or invoke `respond` before the user's answer arrives. Preserve every correlation field in the response, and never reuse an answer file from another request. -- For `permission`, apply the outer Agent's own equivalent permission policy. If the same operation would proceed without asking when invoked directly by the outer Agent, respond `allow_once`; if that policy would deny, respond `deny`; otherwise ask the user. iac-code has already applied its own allow/deny rules, and the outer Agent must not override an iac-code denial. Base the decision on `title`, `purpose`, `effect`, `target`, `isReadOnly`, `deploymentSummary`, and `safeSummary`; do not infer safety from the internal `toolName` alone. When asking about deployment, show the provided plan, region, stack, template, total price, and per-resource prices without exposing raw tool input. -- For `ask_user_question`, present the current prompt and options without inventing a second question. Accept a listed option. Accept free text only when `allowFreeText` is `true`; when present, show `freeTextPrompt` with the input. -- For `candidate_selection`, first present every option's `summary`, render `architectureDiagram` as Mermaid when present, and show `totalMonthlyCost` plus `costItems`. Do not invent missing details or replace these prices with a rough estimate. Then return the selected candidate ID/index requested by the envelope. +- For `permission`, always ask the user to choose one of the returned actions, including for read-only or apparently safe operations. The original request and the outer Agent's permission policy do not authorize an iac-code permission boundary. iac-code has already applied its own allow/deny rules, and the outer Agent must not override an iac-code denial. Present `title`, `purpose`, `effect`, `target`, `isReadOnly`, `deploymentSummary`, and `safeSummary`; do not expose raw tool input or infer safety from the internal `toolName` alone. +- For `ask_user_question`, present the current prompt and options without inventing a second question, then wait for the answer. Accept a listed option. Accept free text only when `allowFreeText` is `true`; when present, show `freeTextPrompt` with the input. +- For `candidate_selection`, present every option's `summary`, render `architectureDiagram` as Mermaid when present, and show `totalMonthlyCost` plus `costItems`, then ask the user to select one. Ask even when there is only one candidate. Do not invent missing details, replace these prices with a rough estimate, or choose on the user's behalf. Return only the candidate ID/index selected by the user. +- For `deployment_confirmation`, present `solutionSummary`, `templateUrl`, the quote or explicit quote failure in `cost`, `effectiveDeploymentParameters`, `parameterOverrides`, `previewReadyForCreate`, and exactly the actions returned in `options`, then ask the user to select an action. A request to create or deploy infrastructure is not confirmation for this boundary. The bridge derives this bounded display projection directly from the Runtime's existing A2A Pipeline confirmation event; never supplement it from local Session, journal, template, spool, or tool-result files. Never confirm, adjust, reselect, or cancel on the user's behalf, including after a failed quote or Preview. - Bind every user answer only to the current `kind`, `inputId`, `requestTaskId`, and `contextId`. Never reinterpret a resource selection as deployment confirmation or reuse it for a later input. -For an automatically decided permission, respond in one tool call while preserving the current input and tool identities: - -```text -python3 scripts/iac_code.py respond --job-id --input-id --tool-use-id --decision allow_once --follow -``` - -Use `deny` when the outer Agent's policy denies it. For a user question, candidate selection, or permission that was explicitly shown to the user, write the correlated answer as JSON to a UTF-8 file and resume the same job: +After the user answers, write the correlated answer as JSON to a UTF-8 file and resume the same job: - Permission: `{"kind":"permission","requestTaskId":"","contextId":"","inputId":"","toolUseId":"","decision":"allow_once"}` or use `deny`. - Question: `{"kind":"ask_user_question","requestTaskId":"","contextId":"","inputId":"","answer":""}`. - Candidate: `{"kind":"candidate_selection","requestTaskId":"","contextId":"","inputId":"","answer":""}`. +- Deployment confirmation: `{"kind":"deployment_confirmation","requestTaskId":"","contextId":"","inputId":"","action":"","parameterOverrides":{"":""}}`. Allowed actions are `confirm`, `adjust`, `reselect`, and `cancel`; omit `parameterOverrides` when the user did not request an adjustment. ```text python3 scripts/iac_code.py respond --job-id --input-file --follow diff --git a/skills/iac-code/SKILL.md.template b/skills/iac-code/SKILL.md.template index fbaab4eb..1db8a90f 100644 --- a/skills/iac-code/SKILL.md.template +++ b/skills/iac-code/SKILL.md.template @@ -29,6 +29,7 @@ Treat a bridge error returned before job creation as the authoritative outcome f ```json { "channel": "codex", + "pipelineName": "selling_solution_first", "permissionWaitPolicy": { "residentTimeoutSeconds": null, "subPipelineTimeoutSeconds": null, @@ -37,10 +38,10 @@ Treat a bridge error returned before job creation as the authoritative outcome f } ``` - `channel` stores only the channel identifier; the bridge adds the `skill/` prefix before sending it to iac-code. `permissionWaitPolicy` applies only to the temporary A2A server owned by this Skill: `null` timeouts mean unlimited waits, positive finite values set resident/Sub Pipeline limits, and grace is a non-negative finite value. Finite values cannot exceed 10 years; use `null` instead of an arbitrarily large number for an unlimited resident or Sub Pipeline wait. The bridge validates and converts this object into server configuration; it never sends the policy through A2A message metadata. Missing fields use the defaults shown above. The bridge rejects unknown configuration fields. If the file or a field is absent, no corresponding override is applied. Never derive these values from the user's request, ask the user for them, or create, edit, or reveal this install-local configuration during an infrastructure task. + `channel` stores only the channel identifier; the bridge adds the `skill/` prefix before sending it to iac-code. `pipelineName` selects the implementation used only after Pipeline mode is chosen: `selling_solution_first` is the default, while the legacy `selling` flow is used only when this install-local file explicitly selects it. `permissionWaitPolicy` applies only to the temporary A2A server owned by this Skill: `null` timeouts mean unlimited waits, positive finite values set resident/Sub Pipeline limits, and grace is a non-negative finite value. Finite values cannot exceed 10 years; use `null` instead of an arbitrarily large number for an unlimited resident or Sub Pipeline wait. The bridge validates and converts this object into server configuration; it never sends the policy through A2A message metadata. The bridge rejects unknown configuration fields and invalid Pipeline names. If `config.json` or `pipelineName` is absent, Pipeline mode uses `selling_solution_first`; other absent fields keep their existing defaults. Never derive these values from the user's request, ask the user for them, or create, edit, or reveal this install-local configuration during an infrastructure task. - Normal is the default, including concrete resource queries/changes, template work, troubleshooting, and deployment of a clear target. Use `--mode pipeline --pipeline-name selling` only when the user explicitly requests it or the request genuinely needs the fixed candidate-architecture, cost-comparison, plan-confirmation, and deployment flow. Questions, permissions, tool use, or deployment alone do not select Pipeline. When uncertain, use normal. - Start performs a non-secret configuration preflight through the Runtime. An incomplete LLM provider/API Key returns `llm_not_configured` and stops before creating a job. Selling Pipeline also requires complete Alibaba Cloud credentials and otherwise returns `cloud_credentials_not_configured`. Normal mode may continue without cloud credentials for work that does not call cloud APIs; report its preflight warning rather than claiming cloud operations are available. + Normal is the overall default, including concrete resource queries/changes, template work, troubleshooting, and deployment of a clear target. Use `--mode pipeline` only when the user explicitly requests it or the request genuinely needs the candidate-architecture, cost-comparison, plan-confirmation, and deployment flow. Pipeline mode uses solution-first unless the installed configuration explicitly selects legacy selling. Questions, permissions, tool use, or deployment alone do not select Pipeline. When uncertain, use normal. + Start performs a non-secret configuration preflight through the Runtime. An incomplete LLM provider/API Key returns `llm_not_configured` and stops before creating a job. Both supported Pipelines require complete Alibaba Cloud credentials and otherwise return `cloud_credentials_not_configured`. Normal mode may continue without cloud credentials for work that does not call cloud APIs; report its preflight warning rather than claiming cloud operations are available. 3. `--follow` consumes the event stream until the next parent/candidate step boundary, permission, user question, candidate selection, `turn_completed`, or terminal state. It writes every parent `step_started`/`step_completed`/`step_failed` and candidate `candidate_step_started`/`candidate_step_completed`/`candidate_step_failed` boundary plus low-frequency bounded heartbeats to stderr; stdout contains one bounded JSON result. A boundary result sets `boundaryReached: true`, `presentationRequired: true`, and provides ready-to-display localized strings in `userUpdates`. Before invoking another tool, emit every `userUpdates` string in a user-visible assistant text block, including the Step 1/2 conclusion already embedded in completed-step updates. Never leave these updates only in reasoning, Bash output, a tool description, or the final summary. After that visible text block, immediately call `follow` again with the returned cursor. Do not treat `boundaryReached` as completion. Do not expand this into raw tool-event or token-delta output. While it is running, do not independently answer the infrastructure task or ask a parallel business question. Only ask the user when the current result contains `inputRequired`. 4. If follow reaches its bounded wait window, call the diagnostic follow command again with the returned cursor: @@ -50,6 +51,7 @@ Treat a bridge error returned before job creation as the authoritative outcome f ``` The recommended wait is 60 seconds and the bridge enforces a 120-second maximum even if a larger value is supplied. + If a result says `state: input-required` but does not contain `inputRequired`, there is no user boundary to answer. Report its `latestText` or error, keep the same job unchanged, and stop. Never call `continue`, repeat `respond`, call `cancel`, or start a replacement job unless the user explicitly requests that action. 5. When `state` is `turn_completed`, treat `finalText` and `artifacts` as the authoritative normal-turn result. When a Pipeline reaches any terminal state, including `completed`, `failed`, `canceled`, or `rejected`, treat `pipelineResult` and `artifacts` as its authoritative result and present its success or failure details directly. If rollback cleanup is pending, the bridge automatically runs a cleanup-only normal task in the same context before returning the Pipeline result; keep following it and handle any returned permission normally. If cleanup is `failed` or `unavailable`, report that manual inspection or retry is required and do not claim it succeeded. Never send a synthetic cleanup prompt or a follow-up merely to retrieve or summarize an existing result. Never recover an answer from Session files, spool files, logs, or raw tool-result files. 6. To send the next natural-language message in the same normal conversation, or after a completed Pipeline has handed the same conversation to normal mode, write it to another workspace prompt file and continue the existing job: @@ -68,24 +70,20 @@ python3 scripts/iac_code.py poll --job-id --cursor --wait-seco ## User input -When `inputRequired` is present, preserve every correlation field in the response. Never reuse an answer file from another request. +Treat every `inputRequired` as a hard user-interaction boundary. Present it through the outer Agent's native user-question or approval UI and stop until the user explicitly answers that specific boundary. If no native UI is available, ask in a visible assistant turn and stop. Never infer, recommend-and-select, or submit an answer from the original infrastructure request, a prior answer, an outer tool-execution approval, a default, or the fact that only one option is available. Do not write an answer file or invoke `respond` before the user's answer arrives. Preserve every correlation field in the response, and never reuse an answer file from another request. -- For `permission`, apply the outer Agent's own equivalent permission policy. If the same operation would proceed without asking when invoked directly by the outer Agent, respond `allow_once`; if that policy would deny, respond `deny`; otherwise ask the user. iac-code has already applied its own allow/deny rules, and the outer Agent must not override an iac-code denial. Base the decision on `title`, `purpose`, `effect`, `target`, `isReadOnly`, `deploymentSummary`, and `safeSummary`; do not infer safety from the internal `toolName` alone. When asking about deployment, show the provided plan, region, stack, template, total price, and per-resource prices without exposing raw tool input. -- For `ask_user_question`, present the current prompt and options without inventing a second question. Accept a listed option. Accept free text only when `allowFreeText` is `true`; when present, show `freeTextPrompt` with the input. -- For `candidate_selection`, first present every option's `summary`, render `architectureDiagram` as Mermaid when present, and show `totalMonthlyCost` plus `costItems`. Do not invent missing details or replace these prices with a rough estimate. Then return the selected candidate ID/index requested by the envelope. +- For `permission`, always ask the user to choose one of the returned actions, including for read-only or apparently safe operations. The original request and the outer Agent's permission policy do not authorize an iac-code permission boundary. iac-code has already applied its own allow/deny rules, and the outer Agent must not override an iac-code denial. Present `title`, `purpose`, `effect`, `target`, `isReadOnly`, `deploymentSummary`, and `safeSummary`; do not expose raw tool input or infer safety from the internal `toolName` alone. +- For `ask_user_question`, present the current prompt and options without inventing a second question, then wait for the answer. Accept a listed option. Accept free text only when `allowFreeText` is `true`; when present, show `freeTextPrompt` with the input. +- For `candidate_selection`, present every option's `summary`, render `architectureDiagram` as Mermaid when present, and show `totalMonthlyCost` plus `costItems`, then ask the user to select one. Ask even when there is only one candidate. Do not invent missing details, replace these prices with a rough estimate, or choose on the user's behalf. Return only the candidate ID/index selected by the user. +- For `deployment_confirmation`, present `solutionSummary`, `templateUrl`, the quote or explicit quote failure in `cost`, `effectiveDeploymentParameters`, `parameterOverrides`, `previewReadyForCreate`, and exactly the actions returned in `options`, then ask the user to select an action. A request to create or deploy infrastructure is not confirmation for this boundary. The bridge derives this bounded display projection directly from the Runtime's existing A2A Pipeline confirmation event; never supplement it from local Session, journal, template, spool, or tool-result files. Never confirm, adjust, reselect, or cancel on the user's behalf, including after a failed quote or Preview. - Bind every user answer only to the current `kind`, `inputId`, `requestTaskId`, and `contextId`. Never reinterpret a resource selection as deployment confirmation or reuse it for a later input. -For an automatically decided permission, respond in one tool call while preserving the current input and tool identities: - -```text -python3 scripts/iac_code.py respond --job-id --input-id --tool-use-id --decision allow_once --follow -``` - -Use `deny` when the outer Agent's policy denies it. For a user question, candidate selection, or permission that was explicitly shown to the user, write the correlated answer as JSON to a UTF-8 file and resume the same job: +After the user answers, write the correlated answer as JSON to a UTF-8 file and resume the same job: - Permission: `{"kind":"permission","requestTaskId":"","contextId":"","inputId":"","toolUseId":"","decision":"allow_once"}` or use `deny`. - Question: `{"kind":"ask_user_question","requestTaskId":"","contextId":"","inputId":"","answer":""}`. - Candidate: `{"kind":"candidate_selection","requestTaskId":"","contextId":"","inputId":"","answer":""}`. +- Deployment confirmation: `{"kind":"deployment_confirmation","requestTaskId":"","contextId":"","inputId":"","action":"","parameterOverrides":{"":""}}`. Allowed actions are `confirm`, `adjust`, `reselect`, and `cancel`; omit `parameterOverrides` when the user did not request an adjustment. ```text python3 scripts/iac_code.py respond --job-id --input-file --follow diff --git a/skills/iac-code/agents/openai.yaml b/skills/iac-code/agents/openai.yaml index 9e9a023f..91ee3aa8 100644 --- a/skills/iac-code/agents/openai.yaml +++ b/skills/iac-code/agents/openai.yaml @@ -1,4 +1,4 @@ interface: display_name: "iac-code" short_description: "Design and operate Alibaba Cloud infrastructure" - default_prompt: "Use $iac-code for this Alibaba Cloud infrastructure task and follow one job through real interaction boundaries. Invoke the packaged bridge before any alternative tool and fail closed if the bridge rejects the host. Default to normal; use selling only when explicitly requested or when candidate architectures, cost comparison, plan confirmation, and deployment are all needed." + default_prompt: "Use $iac-code for this Alibaba Cloud infrastructure task and follow one job through real interaction boundaries. Invoke the packaged bridge before any alternative tool and fail closed if the bridge rejects the host. Never answer an iac-code input boundary on the user's behalf. Default to normal; when Pipeline is warranted, use its solution-first default unless the installed configuration explicitly selects legacy selling." diff --git a/skills/iac-code/scripts/iac_code.py b/skills/iac-code/scripts/iac_code.py index 61d05275..bae9bc4d 100644 --- a/skills/iac-code/scripts/iac_code.py +++ b/skills/iac-code/scripts/iac_code.py @@ -7,6 +7,7 @@ import argparse import contextlib +import copy import ctypes import errno import hashlib @@ -18,6 +19,7 @@ import re import secrets import shutil +import signal import socket import stat import subprocess @@ -83,8 +85,11 @@ } INPUT_STATES = {"input-required", "task_state_input_required"} TURN_COMPLETED_STATE = "turn_completed" -PIPELINE_RESULT_FIELDS = {"selling": "deployment"} -PIPELINE_NORMAL_HANDOFFS = {"selling"} +SESSION_BACKUP_NOT_READY_CODE = "SESSION_BACKUP_NOT_READY" +DEFAULT_PIPELINE_NAME = "selling_solution_first" +SUPPORTED_PIPELINE_NAMES = {"selling", "selling_solution_first"} +PIPELINE_RESULT_FIELDS = {"selling": "deployment", "selling_solution_first": "deployment"} +PIPELINE_NORMAL_HANDOFFS = {"selling", "selling_solution_first"} STEP_BOUNDARY_EVENT_TYPES = { "step_started", "step_completed", @@ -105,6 +110,9 @@ r"(?i)(authorization\s*:\s*bearer\s+|access[_-]?key[_-]?(?:secret|id)?\s*[=:]\s*|" r"api[_-]?key\s*[=:]\s*|token\s*[=:]\s*|password\s*[=:]\s*)([^\s,;]+)" ) +_SENSITIVE_KEY_PATTERN = re.compile( + r"(?i)(password|passwd|secret|token|credential|access.?key|private.?key|client.?secret)" +) def _skill_user_agent(): @@ -139,6 +147,28 @@ def payload(self): return value +def _session_backup_not_ready_error(value): + if not isinstance(value, dict): + return None + candidates = [value] + result = value.get("result") + if isinstance(result, dict): + candidates.append(result) + for candidate in candidates: + raw_error = candidate.get("error") + if not isinstance(raw_error, dict): + continue + data = raw_error.get("data") if isinstance(raw_error.get("data"), dict) else {} + code = data.get("code") or raw_error.get("code") or raw_error.get("Code") + if code != SESSION_BACKUP_NOT_READY_CODE: + continue + message = raw_error.get("message") or raw_error.get("Message") + if not isinstance(message, str) or not message: + message = "Session backup is still synchronizing. Retry after 3 seconds." + return BridgeError(SESSION_BACKUP_NOT_READY_CODE, _sanitize_text(message, 2000), True) + return None + + def _json_bytes(value): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") @@ -946,6 +976,7 @@ def _runtime_matches(record, mode, pipeline_name, target, permission_wait_policy "target": target, "mode": mode, "pipelineName": pipeline_name or "", + "handoffToolConfirmation": True, "permissionWaitPolicy": permission_wait_policy, } if any(record.get(key) != value for key, value in expected.items()) or not _pid_alive(record.get("pid")): @@ -963,6 +994,14 @@ def _runtime_matches(record, mode, pipeline_name, target, permission_wait_policy return card.get("version") == IAC_CODE_VERSION +def _valid_job_pipeline_identity(mode, pipeline_name): + if mode == "normal": + return pipeline_name == "" + if mode == "pipeline": + return pipeline_name in SUPPORTED_PIPELINE_NAMES + return False + + def _runtime_record_for_job(job): record_path = job.get("runtimeRecord") target = job.get("target") @@ -972,9 +1011,7 @@ def _runtime_record_for_job(job): generation = job.get("runtimeGeneration") if not all(isinstance(value, str) and value for value in (record_path, target, mode, workspace, generation)): raise BridgeError("runtime_identity_mismatch", "The Skill job runtime identity is incomplete.") - if mode not in {"normal", "pipeline"} or not isinstance(pipeline_name, str): - raise BridgeError("runtime_identity_mismatch", "The Skill job Pipeline identity is invalid.") - if (mode == "pipeline") != bool(pipeline_name): + if not isinstance(pipeline_name, str) or not _valid_job_pipeline_identity(mode, pipeline_name): raise BridgeError("runtime_identity_mismatch", "The Skill job mode and Pipeline identity do not match.") record = _load_json(pathlib.Path(record_path), "runtime_identity_mismatch") if record.get("generation") != job.get("runtimeGeneration"): @@ -1009,6 +1046,40 @@ def _stop_spawned_process(process): process.wait(timeout=RUNTIME_STOP_TIMEOUT) +def _recorded_runtime_is_ours(record): + pid = record.get("pid") + port = record.get("port") + token = record.get("token") + if not _pid_alive(pid) or not isinstance(port, int) or not isinstance(token, str) or not token: + return False + try: + health = _http_json("http://127.0.0.1:{}/health".format(port), token, timeout=2) + except BridgeError: + return False + return health.get("status") == "healthy" and health.get("version") == record.get("iacCodeVersion") + + +def _stop_recorded_runtime(record): + pid = record.get("pid") + if not _pid_alive(pid): + return True + if not _recorded_runtime_is_ours(record): + return False + with contextlib.suppress(OSError): + os.kill(pid, signal.SIGTERM) + deadline = time.monotonic() + RUNTIME_STOP_TIMEOUT + while _pid_alive(pid) and time.monotonic() < deadline: + time.sleep(0.05) + if _pid_alive(pid): + force_signal = signal.SIGTERM if os.name == "nt" else signal.SIGKILL + with contextlib.suppress(OSError): + os.kill(pid, force_signal) + deadline = time.monotonic() + RUNTIME_STOP_TIMEOUT + while _pid_alive(pid) and time.monotonic() < deadline: + time.sleep(0.05) + return not _pid_alive(pid) + + def _remove_runtime_record(record_path, generation): if not record_path.is_file(): return @@ -1026,11 +1097,27 @@ def ensure_server(executable, artifact, mode, pipeline_name, permission_wait_pol _secure_directory(root) record_path = root / "runtime.json" with InstallLock(root / ".runtime.lock", timeout=10): + stale_record = None if record_path.is_file(): - with contextlib.suppress(BridgeError): + try: record = _load_json(record_path, "runtime_identity_mismatch") - if _runtime_matches(record, mode, pipeline_name, artifact["target"], permission_wait_policy): - return record + except BridgeError as exc: + raise BridgeError( + "runtime_start_failed", + "The existing local iac-code A2A runtime record could not be verified.", + True, + ) from exc + if _runtime_matches(record, mode, pipeline_name, artifact["target"], permission_wait_policy): + return record + stale_record = record + if stale_record is not None: + if not _stop_recorded_runtime(stale_record): + raise BridgeError( + "runtime_start_failed", + "The previous local iac-code A2A runtime could not be stopped safely.", + True, + ) + _remove_runtime_record(record_path, stale_record.get("generation")) token = secrets.token_urlsafe(32) port = _free_port() generation = uuid.uuid4().hex @@ -1061,6 +1148,7 @@ def ensure_server(executable, artifact, mode, pipeline_name, permission_wait_pol environment.pop("IACCODE_A2A_ALLOWED_CWDS", None) environment["IAC_CODE_A2A_TRUST_REQUEST_CWD"] = "1" environment["IAC_CODE_SKILL_RUNTIME_GENERATION"] = generation + environment["IAC_CODE_HANDOFF_USE_TOOL_CONFIRMATION"] = "1" if pipeline_name: environment["IAC_CODE_PIPELINE_NAME"] = pipeline_name else: @@ -1099,6 +1187,7 @@ def ensure_server(executable, artifact, mode, pipeline_name, permission_wait_pol "generation": generation, "mode": mode, "pipelineName": pipeline_name or "", + "handoffToolConfirmation": True, "permissionWaitPolicy": permission_wait_policy, "pid": process.pid, "port": port, @@ -1246,7 +1335,16 @@ def _safe_input_envelope(value): if not isinstance(value, dict): return None kind = value.get("kind") - common = {"schemaVersion", "kind", "requestTaskId", "contextId", "inputId", "prompt", "options", "required"} + common = { + "schemaVersion", + "kind", + "requestTaskId", + "contextId", + "inputId", + "prompt", + "options", + "required", + } if kind == "permission": common.update( { @@ -1264,7 +1362,20 @@ def _safe_input_envelope(value): ) elif kind == "ask_user_question": common.update({"allowFreeText", "freeTextPrompt"}) - elif kind not in {"ask_user_question", "candidate_selection"}: + elif kind == "candidate_selection": + pass + elif kind == "deployment_confirmation": + common.update( + { + "solutionSummary", + "templateUrl", + "cost", + "effectiveDeploymentParameters", + "parameterOverrides", + "previewReadyForCreate", + } + ) + else: return None projected = {key: value[key] for key in common if key in value} projected["prompt"] = _sanitize_text(projected.get("prompt"), 600) @@ -1283,16 +1394,40 @@ def _safe_input_envelope(value): projected["language"] = _sanitize_text(projected["language"], 10) if "deploymentSummary" in projected: projected["deploymentSummary"] = _safe_deployment_summary(projected["deploymentSummary"]) + for key, maximum in (("solutionSummary", 1200), ("templateUrl", 1000)): + if key in projected: + projected[key] = _sanitize_text(projected[key], maximum) + if "previewReadyForCreate" in projected: + projected["previewReadyForCreate"] = projected["previewReadyForCreate"] is True + for key in ("cost", "effectiveDeploymentParameters", "parameterOverrides"): + if key in projected: + projected[key] = _safe_display_value(key, projected[key]) options = projected.get("options") if isinstance(options, list): safe_options = [] for item in options[:20]: - if not isinstance(item, dict) or not isinstance(item.get("id"), str): + if not isinstance(item, dict): continue + if kind == "deployment_confirmation": + action = item.get("action") + if action not in {"confirm", "adjust", "reselect", "cancel"}: + continue + option_id = item.get("id") or action + option_label = item.get("label") or item.get("name") or action + else: + option_id = item.get("id") + option_label = item.get("label") + if not isinstance(option_id, str): + continue safe_item = { - "id": _sanitize_text(item.get("id"), 120), - "label": _sanitize_text(item.get("label"), 200), + "id": _sanitize_text(option_id, 120), + "label": _sanitize_text(option_label, 200), } + if kind == "deployment_confirmation": + safe_item["action"] = action + summary = item.get("summary") or item.get("description") + if isinstance(summary, str) and summary: + safe_item["summary"] = _sanitize_text(summary, 600) if kind == "candidate_selection": for key, maximum in ( ("summary", 600), @@ -1321,6 +1456,36 @@ def _safe_input_envelope(value): return projected +def _safe_display_value(key, value, depth=0, budget=None): + """Bound and filter an existing Runtime confirmation value for Skill display.""" + + if budget is None: + budget = {"nodes": 240} + if budget["nodes"] <= 0 or depth > 6: + return "" + budget["nodes"] -= 1 + if _SENSITIVE_KEY_PATTERN.search(str(key)): + return "" + if isinstance(value, dict): + result = {} + for index, (child_key, child_value) in enumerate(value.items()): + if index >= 60 or budget["nodes"] <= 0: + result["_truncated"] = True + break + safe_key = _sanitize_text(str(child_key), 120) + result[safe_key] = _safe_display_value(safe_key, child_value, depth + 1, budget) + return result + if isinstance(value, list): + return [_safe_display_value(key, item, depth + 1, budget) for item in value[:60]] + if isinstance(value, str): + return _sanitize_text(value, 600) + if value is None or isinstance(value, (bool, int)): + return value + if isinstance(value, float): + return value if math.isfinite(value) else None + return _sanitize_text(str(value), 200) + + def _safe_deployment_summary(value): if not isinstance(value, dict): return None @@ -1633,15 +1798,14 @@ def _safe_step_conclusion_summary(step_id, conclusion_field, value): def _bounded_input_projection(projection): """Keep an input-required control event answerable within the spool/poll budget.""" - bounded = dict(projection) + bounded = copy.deepcopy(projection) envelope = bounded.get("inputRequired") if not isinstance(envelope, dict): return bounded - envelope = dict(envelope) - options = envelope.get("options") - if isinstance(options, list): - envelope["options"] = [dict(item) for item in options if isinstance(item, dict)] - bounded["inputRequired"] = envelope + if envelope.get("kind") == "deployment_confirmation": + return _bounded_deployment_confirmation_projection(bounded, projection) + if len(_json_bytes(bounded)) > MAX_INPUT_PROJECTION_BYTES: + bounded["trimmed"] = True def shrink(key, minimum): value = envelope.get(key) @@ -1701,11 +1865,145 @@ def shrink(key, minimum): "a2a_transport_failed", "The A2A input request exceeded the bounded Skill protocol.", ) - if bounded != projection: + return bounded + + +def _bounded_deployment_confirmation_projection(bounded, original): + del original + envelope = bounded["inputRequired"] + if len(_json_bytes(bounded)) > MAX_INPUT_PROJECTION_BYTES: bounded["trimmed"] = True + + def shrink_text(key, minimum): + value = envelope.get(key) + if not isinstance(value, str) or len(value) <= minimum: + return False + envelope[key] = value[: max(minimum, len(value) // 2)] + return True + + while len(_json_bytes(bounded)) > MAX_INPUT_PROJECTION_BYTES: + changed = shrink_text("solutionSummary", 120) + changed = shrink_text("templateUrl", 80) or changed + changed = shrink_text("prompt", 100) or changed + for option in envelope.get("options", []): + if not isinstance(option, dict): + continue + for key, minimum in (("summary", 40), ("label", 16)): + value = option.get(key) + if isinstance(value, str) and len(value) > minimum: + option[key] = value[: max(minimum, len(value) // 2)] + changed = True + if not changed: + break + + if len(_json_bytes(bounded)) > MAX_INPUT_PROJECTION_BYTES: + cost = envelope.get("cost") + if isinstance(cost, dict): + summary_keys = { + "status", + "quoteStatus", + "quote_status", + "currency", + "totalMonthlyCost", + "total_monthly_cost", + "error", + } + envelope["cost"] = {key: cost[key] for key in cost if key in summary_keys} + + for key in ("effectiveDeploymentParameters", "parameterOverrides"): + values = envelope.get(key) + if not isinstance(values, dict): + continue + while len(_json_bytes(bounded)) > MAX_INPUT_PROJECTION_BYTES and values: + values.pop(next(reversed(values))) + + if len(_json_bytes(bounded)) > MAX_INPUT_PROJECTION_BYTES: + minimal = { + key: envelope[key] + for key in ( + "schemaVersion", + "kind", + "requestTaskId", + "contextId", + "inputId", + "prompt", + "previewReadyForCreate", + ) + if key in envelope + } + minimal["options"] = [ + {key: option[key] for key in ("id", "label", "action") if key in option} + for option in envelope.get("options", []) + if isinstance(option, dict) and option.get("action") in {"confirm", "adjust", "reselect", "cancel"} + ] + bounded["inputRequired"] = minimal + + if len(_json_bytes(bounded)) > MAX_INPUT_PROJECTION_BYTES: + raise BridgeError( + "a2a_transport_failed", + "The A2A deployment confirmation exceeded the minimum bounded protocol.", + ) return bounded +def _deployment_confirmation_from_pipeline_metadata(metadata, task_id, context_id, state): + """Adapt the Runtime's existing Pipeline envelope to the Skill input contract.""" + + if state != "input-required" or not isinstance(metadata, dict): + return None + pipeline_values = [] + batch = metadata.get("pipelineBatch") + if isinstance(batch, dict) and isinstance(batch.get("events"), list): + pipeline_values = batch["events"] + elif isinstance(metadata.get("pipeline"), dict): + pipeline_values = [metadata["pipeline"]] + for envelope in reversed(pipeline_values): + if not isinstance(envelope, dict): + continue + raw_input = envelope.get("input") + if not isinstance(raw_input, dict) and envelope.get("eventType") == "input_required": + raw_input = envelope.get("data") + if not isinstance(raw_input, dict) or raw_input.get("kind") != "deployment_confirmation": + continue + event_task_id = envelope.get("taskId") + event_context_id = envelope.get("contextId") + request_task_id = event_task_id if isinstance(event_task_id, str) and event_task_id else task_id + request_context_id = event_context_id if isinstance(event_context_id, str) and event_context_id else context_id + if not request_task_id or not request_context_id: + return None + input_id = raw_input.get("inputId") or raw_input.get("input_id") + if not isinstance(input_id, str) or not input_id: + event_id = envelope.get("eventId") + input_id = "input-{}".format(event_id if isinstance(event_id, str) and event_id else request_task_id) + normalized = { + "schemaVersion": 1, + "kind": "deployment_confirmation", + "requestTaskId": request_task_id, + "contextId": request_context_id, + "inputId": input_id, + "prompt": raw_input.get("prompt", ""), + "solutionSummary": raw_input.get("solutionSummary", raw_input.get("solution_summary", "")), + "templateUrl": raw_input.get("templateUrl", raw_input.get("template_url", "")), + "cost": raw_input.get("cost", {}), + "effectiveDeploymentParameters": raw_input.get( + "effectiveDeploymentParameters", + raw_input.get("effective_deployment_parameters", {}), + ), + "parameterOverrides": raw_input.get( + "parameterOverrides", + raw_input.get("parameter_overrides", {}), + ), + "previewReadyForCreate": raw_input.get( + "previewReadyForCreate", + raw_input.get("preview_ready_for_create", False), + ), + "options": raw_input.get("options", []), + "required": True, + } + return _safe_input_envelope(normalized) + return None + + def project_frame(frame): result = frame.get("result") if isinstance(frame, dict) else None if not isinstance(result, dict): @@ -1722,6 +2020,8 @@ def project_frame(frame): if state: projected["state"] = state input_value = _safe_input_envelope(metadata.get("input")) + if input_value is None: + input_value = _deployment_confirmation_from_pipeline_metadata(metadata, task_id, context_id, state) if input_value is None and state == "working" and isinstance(metadata.get("pendingPermissions"), list): input_value = next( ( @@ -1825,6 +2125,9 @@ def project_frame(frame): "status": item.get("status"), "sequence": item.get("sequence"), } + pipeline_name = item.get("pipelineName") + if pipeline_name in SUPPORTED_PIPELINE_NAMES: + milestone["pipelineName"] = pipeline_name for key in ("step", "parentStep", "candidate", "candidateStep"): value = item.get(key) if isinstance(value, dict): @@ -2121,6 +2424,7 @@ def _append_projection(job_id, projection, turn_text=""): job["finalTextComplete"] = original.get("finalTextComplete") is True job["finalArtifacts"] = _deduplicated_artifacts(list(original.get("artifacts") or [])) job["turnCompletedAt"] = int(time.time()) + job.pop("assistantFinalReceived", None) job.pop("inputRequired", None) elif original.get("type") == "terminal": if ( @@ -2210,6 +2514,14 @@ def _complete_normal_turn(job_id): ) +def _complete_normal_turn_after_final(job_id): + job = _load_json(_job_paths(job_id)[1]) + if not _job_uses_normal_conversation(job) or job.get("assistantFinalReceived") is not True: + return False + _complete_normal_turn(job_id) + return True + + def _jsonrpc_payload(method, params): return {"jsonrpc": "2.0", "id": str(uuid.uuid4()), "method": method, "params": params} @@ -2328,9 +2640,27 @@ def _worker_payload(job, prompt=None, response=None, cleanup_only=False): if not isinstance(pending, dict): raise BridgeError("input_response_mismatch", "The job has no pending input.") _validate_response_correlation(response, pending) - answer = response.get("answer") if isinstance(response, dict) else None - if answer is None: - answer = response + if response.get("kind") == "deployment_confirmation": + action = response.get("action") + if action not in {"confirm", "adjust", "reselect", "cancel"}: + raise BridgeError( + "input_response_mismatch", + "Deployment confirmation action must be confirm, adjust, reselect, or cancel.", + ) + overrides = response.get("parameterOverrides") + if overrides is not None and not isinstance(overrides, dict): + raise BridgeError( + "input_response_mismatch", + "Deployment confirmation parameterOverrides must be a JSON object.", + ) + business_response = {"action": action} + if overrides is not None: + business_response["parameter_overrides"] = overrides + answer = json.dumps(business_response, ensure_ascii=False, separators=(",", ":")) + else: + answer = response.get("answer") if isinstance(response, dict) else None + if answer is None: + answer = response message.update( { "taskId": job.get("taskId"), @@ -2392,6 +2722,52 @@ def _remove_job_pending_permission(job, input_id): job.pop("inputRequired", None) +def _restore_retryable_permission_response(job_id, error): + _root, job_path, _spool = _job_paths(job_id) + job = _load_json(job_path) + pending = job.get("permissionResponseInput") + if not isinstance(pending, dict) or pending.get("kind") != "permission": + return False + _append_projection( + job_id, + { + "type": "input-required", + "state": "input-required", + "taskId": pending.get("requestTaskId") or job.get("taskId"), + "contextId": pending.get("contextId") or job.get("contextId"), + "inputRequired": pending, + "error": { + "code": error.code, + "message": error.message, + "retryable": True, + }, + "time": int(time.time()), + }, + ) + return True + + +def _clear_permission_response_input(job_id): + root, job_path, _spool = _job_paths(job_id) + with InstallLock(root / ".job.lock", timeout=10): + job = _load_json(job_path) + if "permissionResponseInput" not in job: + return + job.pop("permissionResponseInput", None) + _atomic_json(job_path, job) + + +def _same_permission_input(left, right): + if not isinstance(left, dict) or not isinstance(right, dict): + return False + if left.get("kind") != "permission" or right.get("kind") != "permission": + return False + return all( + isinstance(left.get(key), str) and left.get(key) == right.get(key) + for key in ("requestTaskId", "contextId", "inputId", "toolUseId") + ) + + def _subscription_after_stream(record, job_id): _root, job_path, _spool = _job_paths(job_id) job = _load_json(job_path) @@ -2405,9 +2781,33 @@ def _subscription_after_stream(record, job_id): payload=_jsonrpc_payload("GetTask", {"id": task_id}), timeout=10, ) + backup_not_ready = _session_backup_not_ready_error(status) + if backup_not_ready is not None: + if _restore_retryable_permission_response(job_id, backup_not_ready): + return None + raise backup_not_ready projection = project_frame(status) + pending_response = job.get("permissionResponseInput") + projected_input = projection.get("inputRequired") + if isinstance(pending_response, dict): + if projection.get("type") == "input-required": + if not _same_permission_input(projected_input, pending_response): + _clear_permission_response_input(job_id) + elif projection.get("state") in INPUT_STATES: + backup_not_ready = BridgeError( + SESSION_BACKUP_NOT_READY_CODE, + "Session backup is still synchronizing. Retry after 3 seconds.", + True, + ) + if _restore_retryable_permission_response(job_id, backup_not_ready): + return None + elif projection.get("type") != "diagnostic": + _clear_permission_response_input(job_id) _append_projection(job_id, projection) - if projection.get("type") in {"terminal", "input-required"}: + if projection.get("type") == "terminal": + return None + if projection.get("type") == "input-required": + _complete_normal_turn_after_final(job_id) return None if projection.get("state") in INPUT_STATES: # A normal turn ends in INPUT_REQUIRED without an envelope to mean that @@ -2422,19 +2822,38 @@ def worker(job_id, request_path): job = _load_json(job_path) record = _runtime_record_for_job(job) payload = _load_json(pathlib.Path(request_path), "input_response_mismatch") + permission_response_pending = isinstance(job.get("permissionResponseInput"), dict) attempts = 0 while attempts < 3: attempts += 1 try: for frame in _stream_jsonrpc(record, payload): + backup_not_ready = _session_backup_not_ready_error(frame) + if backup_not_ready is not None: + if _restore_retryable_permission_response(job_id, backup_not_ready): + return 0 + raise backup_not_ready + if permission_response_pending and isinstance(frame.get("result"), dict): + _clear_permission_response_input(job_id) + permission_response_pending = False projection = project_frame(frame) _append_projection(job_id, projection) - if projection.get("type") == "terminal" or projection.get("type") == "input-required": + if projection.get("type") == "terminal": + return 0 + if projection.get("type") == "input-required": + _complete_normal_turn_after_final(job_id) return 0 payload = _subscription_after_stream(record, job_id) if payload is None: return 0 - except (BridgeError, ValueError, UnicodeDecodeError): + except (BridgeError, ValueError, UnicodeDecodeError) as exc: + if ( + isinstance(exc, BridgeError) + and exc.code == SESSION_BACKUP_NOT_READY_CODE + and exc.retryable + and _restore_retryable_permission_response(job_id, exc) + ): + return 0 job = _load_json(job_path) task_id = job.get("taskId") if not isinstance(task_id, str): @@ -2591,7 +3010,11 @@ def _skill_config(): try: encoded = config_path.read_bytes() except FileNotFoundError: - return {"channel": None, "permissionWaitPolicy": None} + return { + "channel": None, + "pipelineName": DEFAULT_PIPELINE_NAME, + "permissionWaitPolicy": None, + } except OSError as exc: raise BridgeError("skill_configuration_invalid", "The installed Skill config could not be read.") from exc if len(encoded) > MAX_SKILL_CONFIG_BYTES: @@ -2602,14 +3025,21 @@ def _skill_config(): raise BridgeError("skill_configuration_invalid", "The installed Skill config is not valid UTF-8 JSON.") from exc if not isinstance(config, dict): raise BridgeError("skill_configuration_invalid", "The installed Skill config must be a JSON object.") - unknown = sorted(str(key) for key in config if key not in {"channel", "permissionWaitPolicy"}) + unknown = sorted(str(key) for key in config if key not in {"channel", "pipelineName", "permissionWaitPolicy"}) if unknown: raise BridgeError( "skill_configuration_invalid", "Unknown installed Skill config fields: {}.".format(", ".join(unknown)), ) + pipeline_name = config.get("pipelineName", DEFAULT_PIPELINE_NAME) + if not isinstance(pipeline_name, str) or pipeline_name not in SUPPORTED_PIPELINE_NAMES: + raise BridgeError( + "skill_configuration_invalid", + "pipelineName must be selling_solution_first or selling.", + ) return { "channel": _normalize_telemetry_channel(config.get("channel")) if config.get("channel") is not None else None, + "pipelineName": pipeline_name, "permissionWaitPolicy": _normalize_permission_wait_policy(config.get("permissionWaitPolicy")), } @@ -2671,14 +3101,23 @@ def start_job(args): if not workspace.is_absolute() or not workspace.exists() or not workspace.is_dir(): raise BridgeError("runtime_identity_mismatch", "The Skill workspace must be an existing absolute directory.") workspace = workspace.resolve() - if args.mode == "pipeline" and not args.pipeline_name: - raise BridgeError("runtime_identity_mismatch", "Pipeline mode requires --pipeline-name.") - if args.mode == "normal" and args.pipeline_name: - raise BridgeError("runtime_identity_mismatch", "Normal mode cannot use a Pipeline name.") prompt = _read_workspace_prompt(workspace, args.prompt_file) preferred_language = _preferred_language(prompt, args.language) _set_output_language(preferred_language) skill_config = _skill_config() + configured_pipeline_name = skill_config["pipelineName"] + cli_pipeline_name = getattr(args, "pipeline_name", "") + if args.mode == "normal": + if cli_pipeline_name: + raise BridgeError("runtime_identity_mismatch", "Normal mode cannot use a Pipeline name.") + pipeline_name = "" + else: + if cli_pipeline_name and cli_pipeline_name != configured_pipeline_name: + raise BridgeError( + "runtime_identity_mismatch", + "The compatibility --pipeline-name value does not match the installed Skill config.", + ) + pipeline_name = configured_pipeline_name channel = skill_config["channel"] permission_wait_policy = skill_config["permissionWaitPolicy"] artifact, executable, cache_hit = ensure_runtime() @@ -2687,12 +3126,12 @@ def start_job(args): executable, artifact, args.mode, - args.pipeline_name, + pipeline_name, permission_wait_policy, ) readiness = _runtime_configuration_readiness( record, - require_cloud=args.mode == "pipeline" and args.pipeline_name == "selling", + require_cloud=args.mode == "pipeline", ) job_id = uuid.uuid4().hex root, job_path, spool = _job_paths(job_id) @@ -2702,7 +3141,7 @@ def start_job(args): os.chmod(str(spool), 0o600) runtime_record = _runtime_record_path( args.mode, - args.pipeline_name, + pipeline_name, artifact["target"], permission_wait_policy, ) @@ -2715,7 +3154,7 @@ def start_job(args): "target": artifact["target"], "mode": args.mode, "conversationMode": args.mode, - "pipelineName": args.pipeline_name or "", + "pipelineName": pipeline_name, "workspace": str(workspace), "preferredLanguage": preferred_language, "runtimeRecord": str(runtime_record), @@ -2759,9 +3198,8 @@ def _ensure_job_runtime(job_id): target = job.get("target") workspace = job.get("workspace") if ( - mode not in {"normal", "pipeline"} - or not isinstance(pipeline_name, str) - or (mode == "pipeline") != bool(pipeline_name) + not isinstance(pipeline_name, str) + or not _valid_job_pipeline_identity(mode, pipeline_name) or not isinstance(target, str) or not target or not isinstance(workspace, str) @@ -2902,6 +3340,8 @@ def _job_result( result[key] = job[key] if job.get("conversationMode") in {"normal", "pipeline"}: result["conversationMode"] = job["conversationMode"] + if job.get("mode") == "pipeline" and job.get("pipelineName") in SUPPORTED_PIPELINE_NAMES: + result["pipelineName"] = job["pipelineName"] if not boundary_reached and isinstance(job.get("cleanup"), dict): result["cleanup"] = job["cleanup"] if not boundary_reached and isinstance(job.get("inputRequired"), dict): @@ -2946,7 +3386,7 @@ def poll_job(args): return _job_result(args.job_id, args.cursor, MAX_POLL_BYTES, preserve_final=False) -def _step_display_name(step_id): +def _step_display_name(step_id, pipeline_name=""): names = { "zh": { "intent_parsing": "理解部署需求", @@ -2957,6 +3397,8 @@ def _step_display_name(step_id): "template_generating": "生成 IaC 模板", "reviewing": "审查 IaC 模板", "cost_estimating": "估算方案成本", + "solution_planning_and_selection": "规划并选择架构方案", + "materialize_selected_candidate": "生成并确认部署方案", }, "en": { "intent_parsing": "understand deployment requirements", @@ -2967,9 +3409,13 @@ def _step_display_name(step_id): "template_generating": "generate the IaC template", "reviewing": "review the IaC template", "cost_estimating": "estimate plan cost", + "solution_planning_and_selection": "plan and select a solution", + "materialize_selected_candidate": "materialize and confirm the selected solution", }, } language_names = names.get(_ACTIVE_LANGUAGE, names["en"]) + if _ACTIVE_LANGUAGE != "zh" and step_id == "deploying" and pipeline_name == "selling_solution_first": + return "deploy the selected solution" if step_id in language_names: return language_names[step_id] return str(step_id or "step").replace("_", " ").replace("-", " ") @@ -2984,7 +3430,7 @@ def _step_progress_detail(milestone): coordinate = milestone.get("parentStep") coordinate = coordinate if isinstance(coordinate, dict) else {} step_id = coordinate.get("id") or coordinate.get("name") - detail = _step_display_name(step_id) if step_id else "" + detail = _step_display_name(step_id, milestone.get("pipelineName")) if step_id else "" index = coordinate.get("index") total = coordinate.get("total") if isinstance(index, int): @@ -3524,6 +3970,8 @@ def respond_job(args): raise BridgeError("input_response_mismatch", "The pending input changed before the response was sent.") current["state"] = "working" current.pop("inputRequired", None) + if response.get("kind") == "permission": + current["permissionResponseInput"] = pending _atomic_json(job_path, current) worker_pid = _spawn_worker(args.job_id, payload) result = { @@ -3641,7 +4089,7 @@ def _parser(): cache_clean.add_argument("--confirm", action="store_true") start = commands.add_parser("start") start.add_argument("--mode", choices=("normal", "pipeline"), default="normal") - start.add_argument("--pipeline-name", default="") + start.add_argument("--pipeline-name", default="", help=argparse.SUPPRESS) start.add_argument("--cwd", required=True) start.add_argument("--prompt-file", required=True) start.add_argument("--language", choices=("auto",) + SUPPORTED_LANGUAGES, default="auto") diff --git a/src/iac_code/a2a/task_store.py b/src/iac_code/a2a/task_store.py index 843f9d8e..eef0281d 100644 --- a/src/iac_code/a2a/task_store.py +++ b/src/iac_code/a2a/task_store.py @@ -152,6 +152,11 @@ def _attach_pending_permissions(self, task: Task) -> None: if not isinstance(iac_code, dict): iac_code = {} metadata["iac_code"] = iac_code + # The A2A SDK merges status metadata into the stored Task. Once an answer moves the + # task out of INPUT_REQUIRED, keeping the old unified input would make GetTask replay + # an already-consumed permission/question and invite a duplicate response. + if task.status.state != TaskState.TASK_STATE_INPUT_REQUIRED: + iac_code.pop("input", None) pending = self._pending_permissions.get(task.id) if pending: iac_code["pendingPermissions"] = [dict(envelope) for envelope in pending] diff --git a/src/iac_code/services/permission_wait.py b/src/iac_code/services/permission_wait.py index e1485bba..93929ff6 100644 --- a/src/iac_code/services/permission_wait.py +++ b/src/iac_code/services/permission_wait.py @@ -357,6 +357,20 @@ async def resolve_permission_execution_identity( credential = AliyunCredentials.load() if credential is None: raise AliyunCallerIdentityUnavailableError("cloud_credentials_unavailable", retryable=False) + if credential.mode == "OAuth": + from iac_code.services.providers.aliyun_oauth import AliyunOAuthError, AliyunOAuthReloginRequired + + try: + credential = AliyunCredentials.refresh_oauth_if_needed(credential) + except AliyunOAuthReloginRequired as exc: + raise AliyunCallerIdentityUnavailableError("oauth_relogin_required", retryable=False) from exc + except AliyunOAuthError as exc: + retryable = ( + exc.status_code is None + or exc.status_code == 429 + or (isinstance(exc.status_code, int) and exc.status_code >= 500) + ) + raise AliyunCallerIdentityUnavailableError("oauth_refresh_failed", retryable=retryable) from exc identity = _cached_caller_identity() if identity is None: identity = await (resolver or AliyunCallerIdentityResolver()).resolve( diff --git a/tests/a2a/test_task_store.py b/tests/a2a/test_task_store.py index b51ae184..9c6bc415 100644 --- a/tests/a2a/test_task_store.py +++ b/tests/a2a/test_task_store.py @@ -11,6 +11,7 @@ from a2a.server.context import ServerCallContext from a2a.types import Artifact, ListTasksRequest, Part, Task, TaskState, TaskStatus from a2a.utils.errors import InvalidParamsError +from google.protobuf.json_format import MessageToDict, ParseDict from google.protobuf.timestamp_pb2 import Timestamp from iac_code.a2a.metrics import NoOpA2AMetrics @@ -1242,6 +1243,39 @@ async def test_save_persists_sdk_state_transitions_and_explicit_task_mirrors(tmp assert snapshot.updated_at == 3 +@pytest.mark.asyncio +async def test_save_clears_consumed_input_metadata_when_task_leaves_input_required() -> None: + store = A2ATaskStore(metrics=NoOpA2AMetrics()) + task = sdk_task("task-1", state=TaskState.TASK_STATE_INPUT_REQUIRED) + ParseDict( + { + "iac_code": { + "input": { + "kind": "permission", + "inputId": "permission-1", + "toolUseId": "tool-1", + } + } + }, + task.metadata, + ) + + await store.save(task) + waiting = await store.get("task-1") + assert waiting is not None + assert MessageToDict(waiting.metadata, preserving_proto_field_name=False)["iac_code"]["input"]["inputId"] == ( + "permission-1" + ) + + task.status.state = TaskState.TASK_STATE_WORKING + await store.save(task) + + resumed = await store.get("task-1") + assert resumed is not None + resumed_metadata = MessageToDict(resumed.metadata, preserving_proto_field_name=False) + assert "input" not in resumed_metadata.get("iac_code", {}) + + @pytest.mark.asyncio async def test_save_retries_failed_shared_task_snapshot_when_sdk_state_is_unchanged(tmp_path) -> None: persistence = FlakyTaskPersistence(tmp_path / "a2a") diff --git a/tests/services/test_permission_wait.py b/tests/services/test_permission_wait.py index 26e28633..2fbbb8c2 100644 --- a/tests/services/test_permission_wait.py +++ b/tests/services/test_permission_wait.py @@ -532,6 +532,43 @@ async def request(_credential, _region_id): assert calls == 2 +@pytest.mark.asyncio +async def test_execution_identity_refreshes_oauth_before_caller_lookup(monkeypatch) -> None: + stale = AliyunCredential( + mode="OAuth", + access_key_id="stale-access-key", + access_key_secret="stale-secret", + sts_token="stale-token", + region_id="cn-hangzhou", + ) + refreshed = AliyunCredential( + mode="OAuth", + access_key_id="refreshed-access-key", + access_key_secret="refreshed-secret", + sts_token="refreshed-token", + region_id="cn-hangzhou", + ) + monkeypatch.setattr(AliyunCredentials, "load", staticmethod(lambda: stale)) + monkeypatch.setattr( + AliyunCredentials, + "refresh_oauth_if_needed", + staticmethod(lambda credential: refreshed if credential is stale else pytest.fail("unexpected credential")), + ) + + async def request(credential, region_id): + assert credential is refreshed + assert region_id == "cn-hangzhou" + return {"IdentityType": "RAMUser", "AccountId": "1001", "UserId": "2002"} + + identity = await resolve_permission_execution_identity( + tool_name="aliyun_api", + tool_input={"region_id": "cn-hangzhou"}, + resolver=AliyunCallerIdentityResolver(request=request), + ) + + assert identity.principal_kind == "ram_user" + + @pytest.mark.asyncio async def test_execution_identity_is_cached_once_per_a2a_request(monkeypatch) -> None: credential_loads = 0 diff --git a/tests/skill_bridge/test_iac_code_bridge.py b/tests/skill_bridge/test_iac_code_bridge.py index 256ab33e..60ffa37b 100644 --- a/tests/skill_bridge/test_iac_code_bridge.py +++ b/tests/skill_bridge/test_iac_code_bridge.py @@ -2,6 +2,7 @@ import argparse import ast +import copy import hashlib import importlib.util import json @@ -99,6 +100,7 @@ def test_bridge_parses_as_python_38_and_uses_only_standard_library_imports() -> ) assert imported_modules <= { "argparse", + "copy", "contextlib", "ctypes", "errno", @@ -113,6 +115,7 @@ def test_bridge_parses_as_python_38_and_uses_only_standard_library_imports() -> "re", "secrets", "shutil", + "signal", "socket", "stat", "subprocess", @@ -297,7 +300,10 @@ def test_projection_marks_authoritative_assistant_final_without_intermediate_tex assert projected["finalTextComplete"] is True -def test_completed_pipeline_projects_deployment_result_and_normal_handoff(monkeypatch, tmp_path: Path) -> None: +@pytest.mark.parametrize("pipeline_name", ["selling", "selling_solution_first"]) +def test_completed_pipeline_projects_deployment_result_and_normal_handoff( + monkeypatch, tmp_path: Path, pipeline_name: str +) -> None: frame = { "result": { "statusUpdate": { @@ -312,6 +318,7 @@ def test_completed_pipeline_projects_deployment_result_and_normal_handoff(monkey "eventType": "step_completed", "status": "completed", "sequence": 20, + "pipelineName": pipeline_name, "data": { "conclusionField": "deployment", "conclusion": { @@ -371,7 +378,7 @@ def test_completed_pipeline_projects_deployment_result_and_normal_handoff(monkey "jobId": job_id, "mode": "pipeline", "conversationMode": "pipeline", - "pipelineName": "selling", + "pipelineName": pipeline_name, "workspace": str(workspace), "state": "working", "turn": 1, @@ -394,6 +401,7 @@ def test_completed_pipeline_projects_deployment_result_and_normal_handoff(monkey result = bridge._job_result(job_id, 0, bridge.MAX_POLL_BYTES, preserve_final=False) assert result["state"] == "completed" assert result["conversationMode"] == "normal" + assert result["pipelineName"] == pipeline_name assert result["pipelineResult"] == projected["pipelineResult"] assert "latestText" not in result assert len(bridge._json_bytes(result)) <= bridge.MAX_POLL_BYTES @@ -850,15 +858,41 @@ def test_job_runtime_identity_rejects_a_replaced_generation(monkeypatch, tmp_pat assert caught.value.code == "runtime_identity_mismatch" +@pytest.mark.parametrize( + ("mode", "pipeline_name"), + [ + ("pipeline", "unknown_pipeline"), + ("pipeline", ""), + ("normal", "selling_solution_first"), + ], +) +def test_job_runtime_identity_rejects_invalid_persisted_pipeline(mode: str, pipeline_name: str, tmp_path: Path) -> None: + job = { + "runtimeRecord": str(tmp_path / "runtime.json"), + "runtimeGeneration": "generation-1", + "mode": mode, + "pipelineName": pipeline_name, + "workspace": str(tmp_path), + "target": "darwin-arm64-macos-cp312", + } + + with pytest.raises(bridge.BridgeError) as caught: + bridge._runtime_record_for_job(job) + + assert caught.value.code == "runtime_identity_mismatch" + + def test_runtime_identity_is_shared_across_workspaces() -> None: target = "darwin-arm64-macos-cp312" normal_record = bridge._runtime_record_path("normal", "", target) same_normal_record = bridge._runtime_record_path("normal", "", target) pipeline_record = bridge._runtime_record_path("pipeline", "selling", target) + solution_first_record = bridge._runtime_record_path("pipeline", "selling_solution_first", target) assert normal_record == same_normal_record assert normal_record != pipeline_record + assert pipeline_record != solution_first_record def test_default_permission_wait_policy_preserves_legacy_runtime_identity() -> None: @@ -897,7 +931,9 @@ def popen(command, **kwargs): _command, kwargs = spawned[0] assert kwargs["cwd"] == str(bridge._runtime_record_path("normal", "", artifact["target"]).parent) assert kwargs["env"]["IAC_CODE_A2A_TRUST_REQUEST_CWD"] == "1" + assert kwargs["env"]["IAC_CODE_HANDOFF_USE_TOOL_CONFIRMATION"] == "1" assert "IACCODE_A2A_ALLOWED_CWDS" not in kwargs["env"] + assert first["handoffToolConfirmation"] is True config = bridge._load_json(Path(first["logPath"]).with_name("a2a.json")) assert config["idle_shutdown_seconds"] == 1800 @@ -944,6 +980,85 @@ def poll(self): assert "permissionWaitPolicy" not in payload["params"]["message"]["metadata"]["iac_code"] +def test_ensure_server_restarts_legacy_behavior_without_changing_durable_root(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) + monkeypatch.setattr(bridge, "_free_port", lambda: 41244) + artifact = {"target": "darwin-arm64-macos-cp312"} + record_path = bridge._runtime_record_path("normal", "", artifact["target"]) + durable_marker = record_path.parent / "a2a" / "existing-task.json" + bridge._secure_directory(durable_marker.parent) + durable_marker.write_text("persisted", encoding="utf-8") + legacy_record = { + "schemaVersion": 2, + "runtimeTag": bridge.RUNTIME_TAG, + "skillVersion": bridge.SKILL_VERSION, + "iacCodeVersion": bridge.IAC_CODE_VERSION, + "target": artifact["target"], + "generation": "legacy-generation", + "mode": "normal", + "pipelineName": "", + "permissionWaitPolicy": None, + "pid": 11111, + "port": 41111, + "token": "legacy-token", + } + bridge._atomic_json(record_path, legacy_record) + stopped = [] + monkeypatch.setattr(bridge, "_stop_recorded_runtime", lambda record: stopped.append(record) or True) + monkeypatch.setattr( + bridge, + "_runtime_matches", + lambda record, *_args: record.get("handoffToolConfirmation") is True, + ) + + class Process: + pid = 22222 + + def poll(self): + return None + + monkeypatch.setattr(bridge.subprocess, "Popen", lambda *_args, **_kwargs: Process()) + + record = bridge.ensure_server(tmp_path / "iac-code", artifact, "normal", "") + + assert stopped == [legacy_record] + assert record["handoffToolConfirmation"] is True + assert Path(record["logPath"]).parent == record_path.parent + assert durable_marker.read_text(encoding="utf-8") == "persisted" + + +def test_ensure_server_rejects_malformed_runtime_record_without_spawning(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) + artifact = {"target": "darwin-arm64-macos-cp312"} + record_path = bridge._runtime_record_path("normal", "", artifact["target"]) + bridge._secure_directory(record_path.parent) + record_path.write_text("{", encoding="utf-8") + monkeypatch.setattr( + bridge.subprocess, + "Popen", + lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not spawn with an unverified record")), + ) + + with pytest.raises(bridge.BridgeError) as caught: + bridge.ensure_server(tmp_path / "iac-code", artifact, "normal", "") + + assert caught.value.code == "runtime_start_failed" + assert caught.value.retryable is True + assert record_path.read_text(encoding="utf-8") == "{" + + +def test_stop_recorded_runtime_refuses_to_kill_an_unverified_live_pid(monkeypatch) -> None: + monkeypatch.setattr(bridge, "_pid_alive", lambda _pid: True) + monkeypatch.setattr(bridge, "_recorded_runtime_is_ours", lambda _record: False) + monkeypatch.setattr( + bridge.os, + "kill", + lambda *_args: (_ for _ in ()).throw(AssertionError("must not kill an unverified process")), + ) + + assert bridge._stop_recorded_runtime({"pid": 12345}) is False + + def test_ensure_server_terminates_failed_spawn_and_removes_record(monkeypatch, tmp_path: Path) -> None: monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setattr(bridge, "_free_port", lambda: 41242) @@ -1209,6 +1324,237 @@ def test_candidate_presentation_survives_bounded_bridge_projection() -> None: assert len(bridge._json_bytes(projection)) <= bridge.MAX_INPUT_PROJECTION_BYTES +@pytest.mark.parametrize("metadata_kind", ["pipeline", "pipelineBatch"]) +def test_deployment_confirmation_projection_adapts_existing_pipeline_envelope( + metadata_kind: str, +) -> None: + secret = "never-leak-deployment-secret" + pipeline_event = { + "eventId": "evt-deployment-1", + "eventType": "input_required", + "pipelineName": "selling_solution_first", + "taskId": "task-1", + "contextId": "ctx-1", + "step": {"id": "materialize_selected_candidate"}, + "data": { + "kind": "deployment_confirmation", + "prompt": "Confirm deployment " + "p" * 900, + "solution_summary": "Safe solution " + "s" * 1800, + "template_url": "templates/selected.yaml", + "cost": { + "quoteStatus": "succeeded", + "totalMonthlyCost": "¥88/月", + "AccessKeySecret": secret, + "resources": [{"name": "ECS", "spec": "x" * 600}] * 20, + }, + "effective_deployment_parameters": { + "VpcName": "demo", + "DatabasePassword": secret, + **{"Parameter{:03d}".format(index): "v" * 100 for index in range(80)}, + }, + "parameter_overrides": {"ApiToken": secret}, + "preview_ready_for_create": True, + "options": [ + {"action": action, "name": action.title()} for action in ("confirm", "adjust", "reselect", "cancel") + ] + + [{"action": "deploy", "name": "Unsafe"}], + }, + } + pipeline_metadata = ( + {"pipeline": pipeline_event} + if metadata_kind == "pipeline" + else { + "pipelineBatch": { + "events": [ + { + "eventId": "evt-step", + "eventType": "step_completed", + "pipelineName": "selling_solution_first", + }, + pipeline_event, + ] + } + } + ) + frame = { + "result": { + "id": "task-1", + "contextId": "ctx-1", + "status": { + "state": "TASK_STATE_INPUT_REQUIRED", + "metadata": { + "iac_code": pipeline_metadata, + }, + }, + } + } + + original = copy.deepcopy(frame) + projection = bridge.project_frame(frame) + pending = projection["inputRequired"] + + assert pending["kind"] == "deployment_confirmation" + assert pending["requestTaskId"] == "task-1" + assert pending["contextId"] == "ctx-1" + assert pending["inputId"] == "input-evt-deployment-1" + assert [option["action"] for option in pending["options"]] == [ + "confirm", + "adjust", + "reselect", + "cancel", + ] + assert projection["trimmed"] is True + assert len(bridge._json_bytes(projection)) <= bridge.MAX_INPUT_PROJECTION_BYTES + assert secret not in json.dumps(projection, ensure_ascii=False) + assert frame == original + + +def test_parameter_only_deployment_confirmation_trim_does_not_mutate_source_and_marks_trimmed() -> None: + projection = { + "type": "input-required", + "inputRequired": { + "schemaVersion": 1, + "kind": "deployment_confirmation", + "requestTaskId": "task-1", + "contextId": "ctx-1", + "inputId": "deployment-parameters-only", + "prompt": "Confirm", + "previewReadyForCreate": True, + "effectiveDeploymentParameters": {"Parameter{:03d}".format(index): "v" * 100 for index in range(80)}, + "parameterOverrides": {}, + "options": [ + {"id": action, "label": action.title(), "action": action} + for action in ("confirm", "adjust", "reselect", "cancel") + ], + }, + } + original = copy.deepcopy(projection) + + bounded = bridge._bounded_input_projection(projection) + + assert projection == original + assert bounded["trimmed"] is True + assert len(bridge._json_bytes(bounded)) <= bridge.MAX_INPUT_PROJECTION_BYTES + pending = bounded["inputRequired"] + assert pending["inputId"] == "deployment-parameters-only" + assert [option["action"] for option in pending["options"]] == ["confirm", "adjust", "reselect", "cancel"] + + +def test_pipeline_result_preserves_existing_bounded_scalar_and_key_behavior() -> None: + secret = "must-not-leak-output-secret" + + projected = bridge._safe_pipeline_result( + { + "status": "success", + "stack_id": "stack-real", + "outputs": { + "ConnectionString": "mysql://admin:@db.internal", + "DatabasePassword": secret, + }, + } + ) + + assert projected == { + "status": "success", + "stack_id": "stack-real", + "outputs": { + "ConnectionString": "mysql://admin:@db.internal", + "DatabasePassword": secret, + }, + } + assert secret in json.dumps(projected) + + +def test_selling_pipeline_result_preserves_legacy_scalar_string_and_key_behavior() -> None: + projected = bridge._safe_pipeline_result( + { + "status": "success", + "outputs": { + "Port": 80, + "Enabled": True, + "TokenEndpoint": "https://sts.example", + }, + } + ) + + assert projected == { + "status": "success", + "outputs": { + "Port": "80", + "Enabled": "True", + "TokenEndpoint": "https://sts.example", + }, + } + + +@pytest.mark.parametrize("action", ["confirm", "adjust", "reselect", "cancel"]) +def test_deployment_confirmation_worker_payload_sends_only_snake_case_business_answer(action: str) -> None: + pending = { + "kind": "deployment_confirmation", + "requestTaskId": "task-1", + "contextId": "ctx-1", + "inputId": "deployment-1", + } + response = { + **pending, + "action": action, + "parameterOverrides": {"InstanceType": "ecs.g8i.large"}, + } + + payload = bridge._worker_payload( + { + "workspace": "/tmp/work", + "taskId": "task-1", + "contextId": "ctx-1", + "inputRequired": pending, + }, + response=response, + ) + + message = payload["params"]["message"] + assert message["taskId"] == "task-1" + assert message["contextId"] == "ctx-1" + assert json.loads(message["parts"][0]["text"]) == { + "action": action, + "parameter_overrides": {"InstanceType": "ecs.g8i.large"}, + } + rendered = json.dumps(message, ensure_ascii=False) + assert "requestTaskId" not in rendered + assert "inputId" not in rendered + + +@pytest.mark.parametrize( + "response_update", + [ + {"action": "deploy"}, + {"action": "confirm", "parameterOverrides": []}, + {"kind": "candidate_selection", "action": "confirm"}, + {"inputId": "deployment-stale", "action": "confirm"}, + ], +) +def test_deployment_confirmation_worker_payload_rejects_invalid_or_stale_response(response_update) -> None: + pending = { + "kind": "deployment_confirmation", + "requestTaskId": "task-1", + "contextId": "ctx-1", + "inputId": "deployment-1", + } + response = {**pending, **response_update} + + with pytest.raises(bridge.BridgeError) as error: + bridge._worker_payload( + { + "workspace": "/tmp/work", + "taskId": "task-1", + "contextId": "ctx-1", + "inputRequired": pending, + }, + response=response, + ) + + assert error.value.code == "input_response_mismatch" + + def test_bridge_detects_language_and_sends_durable_controls_in_a2a_metadata() -> None: assert bridge._preferred_language("请部署一个 VPC", "auto") == "zh" assert bridge._preferred_language("日本語で説明してください", "auto") == "ja" @@ -1307,6 +1653,179 @@ def test_installed_skill_config_rejects_unknown_top_level_fields(monkeypatch, tm bridge._skill_config() +@pytest.mark.parametrize("value", ["", "unknown", 1, None, [], {}]) +def test_installed_skill_pipeline_config_rejects_invalid_values(monkeypatch, tmp_path: Path, value) -> None: + installed_skill = tmp_path / "installed-skill" + installed_skill.mkdir() + (installed_skill / "config.json").write_text(json.dumps({"pipelineName": value}), encoding="utf-8") + monkeypatch.setattr(bridge, "SKILL_ROOT", installed_skill) + + with pytest.raises(bridge.BridgeError) as error: + bridge._skill_config() + + assert error.value.code == "skill_configuration_invalid" + + +@pytest.mark.parametrize( + ("config", "compatibility_name", "expected"), + [ + (None, "", "selling_solution_first"), + ({}, "", "selling_solution_first"), + ({"pipelineName": "selling_solution_first"}, "selling_solution_first", "selling_solution_first"), + ({"pipelineName": "selling"}, "", "selling"), + ({"pipelineName": "selling"}, "selling", "selling"), + ], +) +def test_pipeline_start_uses_only_installed_config_and_persists_effective_identity( + monkeypatch, + tmp_path: Path, + config, + compatibility_name: str, + expected: str, +) -> None: + monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) + installed_skill = tmp_path / "installed-skill" + installed_skill.mkdir() + if config is not None: + (installed_skill / "config.json").write_text(json.dumps(config), encoding="utf-8") + monkeypatch.setattr(bridge, "SKILL_ROOT", installed_skill) + workspace = tmp_path / "workspace" + workspace.mkdir() + prompt = workspace / "prompt.txt" + prompt.write_text("Design and deploy a solution", encoding="utf-8") + monkeypatch.setattr( + bridge, + "ensure_runtime", + lambda: ({"target": "darwin-arm64-macos-cp312"}, tmp_path / "iac-code", True), + ) + server_calls = [] + + def ensure_server(_executable, _artifact, mode, pipeline_name, permission_wait_policy=None): + server_calls.append((mode, pipeline_name, permission_wait_policy)) + return {"port": 41242, "token": "token", "generation": "generation-1"} + + monkeypatch.setattr(bridge, "ensure_server", ensure_server) + readiness_calls = [] + monkeypatch.setattr( + bridge, + "_runtime_configuration_readiness", + lambda _record, require_cloud: readiness_calls.append(require_cloud) or _configuration_readiness(), + ) + spawned = {} + + def spawn(job_id, _payload): + spawned["jobId"] = job_id + return 12345 + + monkeypatch.setattr(bridge, "_spawn_worker", spawn) + monkeypatch.setattr( + bridge, + "_wait_for_task_identity", + lambda job_id, _previous, cursor, worker_pid: { + "ok": True, + "jobId": job_id, + "taskId": "task-1", + "contextId": "ctx-1", + "cursor": cursor, + "state": "working", + "turn": 1, + "workerPid": worker_pid, + "preferredLanguage": "en", + }, + ) + + bridge.start_job( + SimpleNamespace( + cwd=str(workspace), + prompt_file=str(prompt), + language="en", + mode="pipeline", + pipeline_name=compatibility_name, + follow=False, + follow_seconds=0, + ) + ) + + _root, job_path, _spool = bridge._job_paths(spawned["jobId"]) + job = bridge._load_json(job_path) + assert job["pipelineName"] == expected + assert server_calls == [("pipeline", expected, None)] + assert readiness_calls == [True] + + +def test_pipeline_compatibility_flag_cannot_override_installed_config(monkeypatch, tmp_path: Path) -> None: + installed_skill = tmp_path / "installed-skill" + installed_skill.mkdir() + (installed_skill / "config.json").write_text('{"pipelineName":"selling"}', encoding="utf-8") + monkeypatch.setattr(bridge, "SKILL_ROOT", installed_skill) + workspace = tmp_path / "workspace" + workspace.mkdir() + prompt = workspace / "prompt.txt" + prompt.write_text("Deploy", encoding="utf-8") + monkeypatch.setattr(bridge, "ensure_runtime", lambda: pytest.fail("must fail before Runtime startup")) + + with pytest.raises(bridge.BridgeError) as error: + bridge.start_job( + SimpleNamespace( + cwd=str(workspace), + prompt_file=str(prompt), + language="en", + mode="pipeline", + pipeline_name="selling_solution_first", + follow=False, + follow_seconds=0, + ) + ) + + assert error.value.code == "runtime_identity_mismatch" + + +def test_existing_job_recovery_uses_persisted_pipeline_after_config_changes(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) + workspace = tmp_path / "workspace" + workspace.mkdir() + job_id = "9" * 32 + root, job_path, _spool = bridge._job_paths(job_id) + bridge._secure_directory(root) + bridge._atomic_json( + job_path, + { + "runtimeIdentityVersion": 2, + "runtimeTag": bridge.RUNTIME_TAG, + "runtimeGeneration": "old-generation", + "runtimeRecord": str(tmp_path / "old-runtime.json"), + "target": "darwin-arm64-macos-cp312", + "mode": "pipeline", + "pipelineName": "selling", + "workspace": str(workspace), + }, + ) + monkeypatch.setattr( + bridge, + "_runtime_record_for_job", + lambda _job: (_ for _ in ()).throw(bridge.BridgeError("runtime_identity_mismatch", "stale")), + ) + monkeypatch.setattr( + bridge, + "ensure_runtime", + lambda: ({"target": "darwin-arm64-macos-cp312"}, tmp_path / "iac-code", True), + ) + captured = {} + + def ensure_server(_executable, _artifact, mode, pipeline_name, permission_wait_policy=None): + captured.update(mode=mode, pipelineName=pipeline_name, permissionWaitPolicy=permission_wait_policy) + return {"port": 41242, "token": "token", "generation": "new-generation"} + + monkeypatch.setattr(bridge, "ensure_server", ensure_server) + monkeypatch.setattr(bridge, "_skill_config", lambda: pytest.fail("recovery must not re-read config.json")) + + recovered, _record = bridge._ensure_job_runtime(job_id) + + assert captured == {"mode": "pipeline", "pipelineName": "selling", "permissionWaitPolicy": None} + assert recovered["pipelineName"] == "selling" + assert recovered["runtimeGeneration"] == "new-generation" + + def test_job_results_repeat_preferred_language(monkeypatch, tmp_path: Path) -> None: monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) job_id = "8" * 32 @@ -1425,7 +1944,7 @@ def fail_readiness(_record, require_cloud): prompt_file=str(prompt), language="auto", mode="pipeline", - pipeline_name="selling", + pipeline_name="", follow=False, follow_seconds=0, ) @@ -1441,7 +1960,7 @@ def test_start_persists_installed_skill_channel_and_injects_it_into_a2a_metadata monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) installed_skill = tmp_path / "installed-skill" installed_skill.mkdir() - (installed_skill / "config.json").write_text('{"channel":" host "}', encoding="utf-8") + (installed_skill / "config.json").write_text('{"channel":" host ","pipelineName":"selling"}', encoding="utf-8") monkeypatch.setattr(bridge, "SKILL_ROOT", installed_skill) workspace = tmp_path / "workspace" workspace.mkdir() @@ -1502,6 +2021,7 @@ def spawn(job_id, payload): job = bridge._load_json(job_path) metadata = captured["payload"]["params"]["message"]["metadata"]["iac_code"] assert job["channel"] == "skill/host" + assert job["pipelineName"] == "" assert metadata["channel"] == "skill/host" assert result["contextId"] == "context-1" @@ -1552,6 +2072,40 @@ def test_follow_surfaces_every_parent_and_candidate_step_boundary() -> None: assert len({signature for signature, _message in messages}) == len(messages) +def test_solution_first_step_names_are_localized_for_skill_progress() -> None: + try: + bridge._set_output_language("zh") + assert bridge._step_display_name("solution_planning_and_selection") == "规划并选择架构方案" + assert bridge._step_display_name("materialize_selected_candidate") == "生成并确认部署方案" + assert bridge._step_display_name("deploying", "selling") == "部署选定方案" + assert bridge._step_display_name("deploying", "selling_solution_first") == "部署选定方案" + bridge._set_output_language("en") + assert bridge._step_display_name("solution_planning_and_selection") == "plan and select a solution" + assert bridge._step_display_name("materialize_selected_candidate") == ( + "materialize and confirm the selected solution" + ) + assert bridge._step_display_name("deploying", "selling") == "deploy the selected plan" + assert bridge._step_display_name("deploying", "selling_solution_first") == "deploy the selected solution" + assert ( + bridge._step_progress_detail( + {"eventType": "step_started", "pipelineName": "selling", "step": {"id": "deploying"}} + ) + == "deploy the selected plan" + ) + assert ( + bridge._step_progress_detail( + { + "eventType": "step_started", + "pipelineName": "selling_solution_first", + "step": {"id": "deploying"}, + } + ) + == "deploy the selected solution" + ) + finally: + bridge._set_output_language("en") + + def test_projection_keeps_step_boundaries_and_coordinates_ahead_of_recent_tools() -> None: events = [ { @@ -1845,6 +2399,7 @@ def spawn(_job_id, payload): "decision": "allow_once", } assert result["state"] == "working" + assert bridge._load_json(job_path)["permissionResponseInput"] == pending bridge._atomic_json(job_path, {**bridge._load_json(job_path), "inputRequired": pending, "state": "input-required"}) with pytest.raises(bridge.BridgeError, match="correlation fields"): @@ -1943,6 +2498,251 @@ def send(_url, _token, **kwargs): assert job["workerPid"] == 4321 +def test_worker_restores_permission_when_session_backup_is_not_ready(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) + workspace = tmp_path / "workspace" + workspace.mkdir() + job_id = "8" * 32 + root, job_path, spool = bridge._job_paths(job_id) + bridge._secure_directory(root) + spool.touch() + pending = { + "kind": "permission", + "requestTaskId": "task-1", + "contextId": "ctx-1", + "inputId": "permission-1", + "toolUseId": "tool-1", + } + bridge._atomic_json( + job_path, + { + "jobId": job_id, + "mode": "normal", + "conversationMode": "normal", + "workspace": str(workspace), + "state": "working", + "turn": 1, + "taskId": "task-1", + "contextId": "ctx-1", + "permissionResponseInput": pending, + "turnArtifacts": [], + }, + ) + request_path = root / "request.json" + bridge._atomic_json(request_path, {"jsonrpc": "2.0"}) + monkeypatch.setattr(bridge, "_runtime_record_for_job", lambda _job: {}) + monkeypatch.setattr( + bridge, + "_stream_jsonrpc", + lambda *_args: iter( + [ + { + "error": { + "code": -32602, + "message": "Session backup is still synchronizing. Retry after 3 seconds.", + "data": {"code": bridge.SESSION_BACKUP_NOT_READY_CODE, "retryable": True}, + } + } + ] + ), + ) + monkeypatch.setattr( + bridge, + "_subscription_after_stream", + lambda *_args: (_ for _ in ()).throw(AssertionError("must not treat retry as stream completion")), + ) + + assert bridge.worker(job_id, request_path) == 0 + + job = bridge._load_json(job_path) + assert job["state"] == "input-required" + assert job["inputRequired"] == pending + assert job["permissionResponseInput"] == pending + [event] = bridge._read_spool(spool) + assert event["type"] == "input-required" + assert event["inputRequired"] == pending + assert event["error"] == { + "code": bridge.SESSION_BACKUP_NOT_READY_CODE, + "message": "Session backup is still synchronizing. Retry after 3 seconds.", + "retryable": True, + } + + +def test_worker_clears_permission_retry_marker_after_runtime_accepts_response(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) + workspace = tmp_path / "workspace" + workspace.mkdir() + job_id = "9" * 32 + root, job_path, spool = bridge._job_paths(job_id) + bridge._secure_directory(root) + spool.touch() + pending = { + "kind": "permission", + "requestTaskId": "task-1", + "contextId": "ctx-1", + "inputId": "permission-1", + "toolUseId": "tool-1", + } + bridge._atomic_json( + job_path, + { + "jobId": job_id, + "mode": "normal", + "conversationMode": "normal", + "workspace": str(workspace), + "state": "working", + "turn": 1, + "taskId": "task-1", + "contextId": "ctx-1", + "permissionResponseInput": pending, + "turnArtifacts": [], + }, + ) + request_path = root / "request.json" + bridge._atomic_json(request_path, {"jsonrpc": "2.0"}) + monkeypatch.setattr(bridge, "_runtime_record_for_job", lambda _job: {}) + monkeypatch.setattr( + bridge, + "_stream_jsonrpc", + lambda *_args: iter( + [ + { + "result": { + "taskId": "task-1", + "contextId": "ctx-1", + "status": {"state": "TASK_STATE_WORKING"}, + } + } + ] + ), + ) + monkeypatch.setattr(bridge, "_subscription_after_stream", lambda *_args: None) + + assert bridge.worker(job_id, request_path) == 0 + assert "permissionResponseInput" not in bridge._load_json(job_path) + + +@pytest.mark.parametrize("advanced_state", ["completed", "new_input"]) +def test_get_task_fallback_clears_permission_marker_after_task_advances( + monkeypatch, tmp_path: Path, advanced_state: str +) -> None: + monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) + workspace = tmp_path / "workspace" + workspace.mkdir() + job_id = ("a" if advanced_state == "completed" else "b") * 32 + root, job_path, spool = bridge._job_paths(job_id) + bridge._secure_directory(root) + spool.touch() + pending = { + "kind": "permission", + "requestTaskId": "task-1", + "contextId": "ctx-1", + "inputId": "permission-1", + "toolUseId": "tool-1", + } + bridge._atomic_json( + job_path, + { + "jobId": job_id, + "mode": "normal", + "conversationMode": "normal", + "workspace": str(workspace), + "state": "working", + "turn": 1, + "taskId": "task-1", + "contextId": "ctx-1", + "permissionResponseInput": pending, + "turnArtifacts": [], + }, + ) + if advanced_state == "completed": + task = { + "id": "task-1", + "contextId": "ctx-1", + "status": {"state": "TASK_STATE_COMPLETED"}, + } + expected_input = None + else: + expected_input = { + "kind": "ask_user_question", + "requestTaskId": "task-1", + "contextId": "ctx-1", + "inputId": "question-2", + "prompt": "Choose the next action", + "options": [{"id": "continue", "label": "Continue"}], + } + task = { + "id": "task-1", + "contextId": "ctx-1", + "status": {"state": "TASK_STATE_INPUT_REQUIRED"}, + "metadata": {"iac_code": {"input": expected_input}}, + } + monkeypatch.setattr(bridge, "_http_json", lambda *_args, **_kwargs: {"result": {"task": task}}) + + assert bridge._subscription_after_stream({"port": 1, "token": "token"}, job_id) is None + + job = bridge._load_json(job_path) + assert "permissionResponseInput" not in job + if expected_input is None: + assert job["state"] == "completed" + assert "inputRequired" not in job + else: + assert job["state"] == "input-required" + assert job["inputRequired"]["inputId"] == expected_input["inputId"] + + +def test_get_task_fallback_restores_permission_when_input_metadata_is_not_ready(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) + workspace = tmp_path / "workspace" + workspace.mkdir() + job_id = "c" * 32 + root, job_path, spool = bridge._job_paths(job_id) + bridge._secure_directory(root) + spool.touch() + pending = { + "kind": "permission", + "requestTaskId": "task-1", + "contextId": "ctx-1", + "inputId": "permission-1", + "toolUseId": "tool-1", + } + bridge._atomic_json( + job_path, + { + "jobId": job_id, + "mode": "normal", + "conversationMode": "normal", + "workspace": str(workspace), + "state": "working", + "turn": 1, + "taskId": "task-1", + "contextId": "ctx-1", + "permissionResponseInput": pending, + "turnArtifacts": [], + }, + ) + monkeypatch.setattr( + bridge, + "_http_json", + lambda *_args, **_kwargs: { + "result": { + "task": { + "id": "task-1", + "contextId": "ctx-1", + "status": {"state": "TASK_STATE_INPUT_REQUIRED"}, + } + } + }, + ) + + assert bridge._subscription_after_stream({"port": 1, "token": "token"}, job_id) is None + + job = bridge._load_json(job_path) + assert job["state"] == "input-required" + assert job["inputRequired"] == pending + assert job["permissionResponseInput"] == pending + + def test_normal_turn_aggregates_authoritative_result_without_private_session_files(monkeypatch, tmp_path: Path) -> None: monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) workspace = tmp_path / "workspace" @@ -1998,7 +2798,7 @@ def test_normal_turn_aggregates_authoritative_result_without_private_session_fil assert len(bridge._json_bytes(result)) <= bridge.MAX_FOLLOW_BYTES -def test_worker_emits_turn_completed_after_normal_stream_end(monkeypatch, tmp_path: Path) -> None: +def test_worker_emits_turn_completed_when_stale_input_replays_after_normal_final(monkeypatch, tmp_path: Path) -> None: monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(tmp_path / "config")) workspace = tmp_path / "workspace" workspace.mkdir() @@ -2058,6 +2858,21 @@ def stream(_record, _payload): "id": "task-1", "contextId": "ctx-1", "status": {"state": "TASK_STATE_INPUT_REQUIRED"}, + "metadata": { + "iac_code": { + "input": { + "schemaVersion": 1, + "kind": "permission", + "requestTaskId": "task-1", + "contextId": "ctx-1", + "inputId": "permission-consumed", + "toolUseId": "tool-1", + "prompt": "stale permission", + "options": [], + "required": True, + } + } + }, } }, ) @@ -2067,6 +2882,8 @@ def stream(_record, _payload): assert job["state"] == "turn_completed" assert job["finalText"] == "Hello world" assert job["finalTextComplete"] is True + assert "inputRequired" not in job + assert "assistantFinalReceived" not in job def test_long_normal_result_becomes_public_workspace_artifact(monkeypatch, tmp_path: Path) -> None: @@ -2627,7 +3444,7 @@ def test_skill_contract_uses_implicit_trigger_normal_default_and_follow() -> Non assert "the first operational command must invoke the packaged bridge" in skill assert "when the bridge returns `incompatible_host`" in skill assert "do not bypass the bridge with direct cloud calls" in skill - assert "Normal is the default" in skill + assert "Normal is the overall default" in skill assert "candidate-architecture, cost-comparison, plan-confirmation" in skill assert "start --mode normal" in skill and "--follow" in skill assert "python3 scripts/iac_code.py continue" in skill @@ -2637,9 +3454,18 @@ def test_skill_contract_uses_implicit_trigger_normal_default_and_follow() -> Non assert "Pipeline reaches any terminal state" in skill assert "treat `pipelineResult` and `artifacts` as its authoritative result" in skill assert "bridge enforces a 120-second maximum" in skill + assert "`state: input-required` but does not contain `inputRequired`" in skill + assert "keep the same job unchanged" in skill + assert "Never call `continue`, repeat `respond`, call `cancel`, or start a replacement job" in skill assert "never translate Chinese user-visible content into English" in skill assert "Never call `start --mode normal` to continue a completed Pipeline" in skill - assert "apply the outer Agent's own equivalent permission policy" in skill + assert "Treat every `inputRequired` as a hard user-interaction boundary" in skill + assert "Do not write an answer file or invoke `respond` before the user's answer arrives" in skill + assert "always ask the user" in skill + assert "Ask even when there is only one candidate" in skill + assert "A request to create or deploy infrastructure is not confirmation" in skill + assert "For an automatically decided permission" not in skill + assert '"action":"confirm"' not in skill assert "`presentationRequired: true`" in skill assert "ready-to-display localized strings in `userUpdates`" in skill assert "Before invoking another tool, emit every `userUpdates` string" in skill @@ -2649,6 +3475,12 @@ def test_skill_contract_uses_implicit_trigger_normal_default_and_follow() -> Non assert "`llm_not_configured`" in skill assert "`cloud_credentials_not_configured`" in skill assert "optional `config.json` beside this `SKILL.md`" in skill + assert '"pipelineName": "selling_solution_first"' in skill + pipeline_default_contract = ( + "Pipeline mode uses solution-first unless the installed configuration explicitly selects legacy selling" + ) + assert pipeline_default_contract in skill + assert "--pipeline-name" not in skill assert '"permissionWaitPolicy"' in skill assert '"residentTimeoutSeconds"' in skill assert "never sends the policy through A2A message metadata" in skill @@ -2660,8 +3492,11 @@ def test_skill_contract_uses_implicit_trigger_normal_default_and_follow() -> Non assert "session.jsonl" not in skill assert "`pip install" not in skill assert "Default to normal" in agent_metadata + assert "Never answer an iac-code input boundary on the user's behalf" in agent_metadata assert "fail closed if the bridge rejects the host" in agent_metadata - assert "candidate architectures, cost comparison, plan confirmation" in agent_metadata + assert "use its solution-first default unless the installed configuration explicitly selects legacy selling" in ( + agent_metadata + ) def test_parser_exposes_continue_follow_and_diagnostic_poll() -> None: @@ -2687,6 +3522,10 @@ def test_parser_exposes_continue_follow_and_diagnostic_poll() -> None: ) cache_list = parser.parse_args(["cache", "list"]) cache_clean = parser.parse_args(["cache", "clean", "--candidates", "--confirm"]) + pipeline_start = parser.parse_args( + ["start", "--mode", "pipeline", "--cwd", "/workspace", "--prompt-file", "/workspace/prompt.txt"] + ) + subparsers = next(action for action in parser._actions if isinstance(action, argparse._SubParsersAction)) assert continued.follow is True assert followed.cursor == 7 and followed.wait_seconds == 60 @@ -2695,6 +3534,8 @@ def test_parser_exposes_continue_follow_and_diagnostic_poll() -> None: assert permission.decision == "allow_once" assert cache_list.cache_command == "list" assert cache_clean.candidates is True and cache_clean.confirm is True + assert pipeline_start.pipeline_name == "" + assert "--pipeline-name" not in subparsers.choices["start"].format_help() def _cached_runtime(config_root: Path, runtime_tag: str, target: str, content: bytes) -> Path: