feat(extensions): add read-only configuration provider contract

docs/architecture/rfcs/configuration-assembly-provider-v0.md

@@ -0,0 +1,44 @@
+# RFC: Configuration Assembly Provider (v0)
+
+- **RFC status:** Draft
+- **Delivery maturity:** Partial
+- **Authors / owners:** LoopX maintainers
+- **Created:** 2026-09-14
+- **Last normative revision:** 2026-09-14
+- **Implementation baseline:** issue #3800
+- **Related contracts:** [Host Integration Surface v0](../../reference/protocols/host-integration-surface-v0.md), [TypeScript control-plane migration v0](typescript-control-plane-migration-v0.md)
+- **Language mirror:** [中文版](configuration-assembly-provider-v0.zh-CN.md)
+
+## 1. Decision summary
+
+LoopX may call an independently installed configuration-assembly provider across a versioned JSON process boundary. Version 0 is default-off and read-only: only `probe` and `plan` exist. LoopX remains authoritative for goals, Todo, gates, quota, recovery, and accepted writeback. A provider plan, status, or zero exit code never validates task completion or grants launch, confirmation, or write authority.
+
+## 2. Placement
+
+The provider is an optional extension boundary, not a new outcome capability and not a host runtime. Host adapters remain thin under Host Integration Surface v0. The process contract follows the TypeScript migration rule: coarse versioned JSON calls, never in-process imports or copied databases.
+
+## 3. Protocol
+
+Requests contain `schema_version`, `operation`, `operation_id`, and a public-safe `request`. Responses must echo the schema, operation, and operation id. `plan` additionally returns `plan_id`, canonical SHA-256 `plan_digest`, and a JSON plan. Status is exactly `ready`, `unknown`, or `incomplete`; LoopX preserves `unknown` and `incomplete` rather than promoting them.
+
+Only allowlisted response fields enter LoopX readback. Provider stderr and undeclared response fields are discarded. Output is bounded to 64 KiB and execution is timed out. Missing executables, malformed or truncated JSON, oversized output, version drift, identity mismatch, digest mismatch, timeout, and nonzero exit all fail closed as public-safe read failures.
+
+## 4. Feature-off parity
+
+Absent, disabled, incompatible, or failed providers return unavailable/unknown readback and do not alter the existing lifecycle. No result from this protocol sets completion, gate, validation, launch, or writeback state.
+
+## 5. CLI fallback
+
+The stable fallback is a direct JSON request to the configured executable over stdin and one JSON response over stdout. Hosts may wrap the same call but gain no additional authority.
+
+## 6. Non-goals
+
+Version 0 does not approve `apply`, `observe`, `recover`, client launch, Skills/workflow-kit integration, provider-managed confirmation, or provider-owned LoopX state.
+
+## 7. Validation
+
+Focused tests cover feature-off parity; public-field filtering; preservation of incomplete; missing executable; malformed JSON; schema, operation, identity, and digest fencing; timeout; and the absence of completion/validation authority.
+
+## 8. Smallest delivered slice
+
+`loopx.configuration_assembly_provider.invoke_configuration_provider` supplies the provider-neutral read boundary. A synthetic executable fixture exercises the same stdin/stdout contract without coupling the protocol to any particular provider.

docs/architecture/rfcs/configuration-assembly-provider-v0.zh-CN.md

@@ -0,0 +1,44 @@
+# RFC：配置装配 Provider（v0）
+
+- **RFC 状态：** 草案
+- **交付成熟度：** 部分实现
+- **作者 / Owner：** LoopX maintainers
+- **创建日期：** 2026-09-14
+- **最后规范修订：** 2026-09-14
+- **实现基线：** issue #3800
+- **相关契约：** [Host Integration Surface v0](../../reference/protocols/host-integration-surface-v0.md)、[TypeScript control-plane migration v0](typescript-control-plane-migration-v0.zh-CN.md)
+- **语言镜像：** [English](configuration-assembly-provider-v0.md)
+
+## 1. 决策摘要
+
+LoopX 可经版本化 JSON 进程边界调用独立安装的配置装配 provider。v0 默认关闭且只读，仅有 `probe`、`plan`。goal、Todo、gate、quota、recovery 与 accepted writeback 仍由 LoopX 掌权。provider 的 plan、status 或退出码 0 均不得验证任务完成，亦不得授予启动、确认或写权限。
+
+## 2. 放置
+
+此 provider 是可选扩展边界，不是新的 outcome capability，也不是 host runtime。host adapter 仍遵守 Host Integration Surface v0 的薄适配原则。进程契约遵守 TypeScript 迁移规则：粗粒度版本化 JSON 调用，不作进程内 import，不复制数据库。
+
+## 3. 协议
+
+请求含 `schema_version`、`operation`、`operation_id` 与 public-safe `request`。响应必须回显 schema、operation、operation id。`plan` 另含 `plan_id`、规范 JSON 的 SHA-256 `plan_digest` 及 JSON plan。status 仅可为 `ready`、`unknown`、`incomplete`；LoopX 原样保留 `unknown`、`incomplete`，不得提升。
+
+仅白名单响应字段进入 LoopX readback。provider stderr 与未声明字段一律丢弃。输出上限 64 KiB，执行有超时。可执行文件缺失、JSON 畸形或截断、输出超限、版本漂移、身份或 digest 不符、超时、非零退出均按 public-safe 只读失败关闭。
+
+## 4. 关闭态等价
+
+provider 缺失、禁用、不兼容或失败时，仅返回 unavailable/unknown readback，不改变既有 lifecycle。此协议任何结果均不得设置 completion、gate、validation、launch 或 writeback 状态。
+
+## 5. CLI fallback
+
+稳定 fallback 是：向配置的可执行文件 stdin 发送单个 JSON 请求，从 stdout 接收单个 JSON 响应。host 可薄封装同一调用，但不获得额外权限。
+
+## 6. 非目标
+
+v0 不批准 `apply`、`observe`、`recover`、客户端启动、Skills/workflow-kit 集成、provider 管理确认，亦不批准 provider 持有 LoopX 状态。
+
+## 7. 验证
+
+聚焦测试覆盖关闭态等价、公开字段过滤、incomplete 保留、可执行文件缺失、畸形 JSON、schema/operation/身份/digest fence、超时，以及不得产生完成或验证权限。
+
+## 8. 最小交付
+
+`loopx.configuration_assembly_provider.invoke_configuration_provider` 提供 provider-neutral 只读边界。合成可执行 fixture 通过同一 stdin/stdout 契约验证，不把协议绑定到某个 provider。

loopx/configuration_assembly_provider.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from .extensions.process_runtime import run_capped_process
+
+SCHEMA_VERSION = "loopx_configuration_assembly_provider_v0"
+MAX_RESPONSE_BYTES = 64 * 1024
+ALLOWED_OPERATIONS = {"probe", "plan"}
+ALLOWED_STATUSES = {"ready", "unknown", "incomplete"}
+
+
+def _failure(operation: str, operation_id: str, kind: str) -> dict[str, Any]:
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "operation": operation,
+        "operation_id": operation_id,
+        "status": "unknown",
+        "available": False,
+        "failure_kind": kind,
+    }
+
+
+def plan_digest(plan: Mapping[str, Any]) -> str:
+    encoded = json.dumps(dict(plan), sort_keys=True, separators=(",", ":")).encode()
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def invoke_configuration_provider(
+    *,
+    operation: str,
+    operation_id: str,
+    argv: Sequence[str] | None,
+    enabled: bool = False,
+    timeout_seconds: float = 5,
+    request: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Invoke the default-off read-only provider without granting lifecycle authority."""
+    if operation not in ALLOWED_OPERATIONS:
+        raise ValueError("configuration provider operation must be probe or plan")
+    if not enabled:
+        return _failure(operation, operation_id, "disabled")
+    if not argv:
+        return _failure(operation, operation_id, "missing_executable")
+    payload = {
+        "schema_version": SCHEMA_VERSION,
+        "operation": operation,
+        "operation_id": operation_id,
+        "request": dict(request or {}),
+    }
+    try:
+        result = run_capped_process(
+            argv,
+            stdin=json.dumps(payload, separators=(",", ":")).encode(),
+            timeout_seconds=timeout_seconds,
+            output_limit_bytes=MAX_RESPONSE_BYTES,
+        )
+    except OSError:
+        return _failure(operation, operation_id, "missing_executable")
+    if result.failure_kind:
+        return _failure(operation, operation_id, result.failure_kind)
+    if result.returncode != 0:
+        return _failure(operation, operation_id, "provider_failed")
+    try:
+        response = json.loads(result.stdout)
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return _failure(operation, operation_id, "malformed_response")
+    if not isinstance(response, dict):
+        return _failure(operation, operation_id, "malformed_response")
+    if response.get("schema_version") != SCHEMA_VERSION:
+        return _failure(operation, operation_id, "incompatible_schema")
+    if response.get("operation") != operation or response.get("operation_id") != operation_id:
+        return _failure(operation, operation_id, "identity_mismatch")
+    status = response.get("status")
+    if status not in ALLOWED_STATUSES:
+        return _failure(operation, operation_id, "invalid_status")
+    public = {
+        "schema_version": SCHEMA_VERSION,
+        "operation": operation,
+        "operation_id": operation_id,
+        "status": status,
+        "available": True,
+    }
+    if operation == "plan":
+        plan = response.get("plan")
+        if not isinstance(plan, dict) or not isinstance(response.get("plan_id"), str):
+            return _failure(operation, operation_id, "invalid_plan")
+        if response.get("plan_digest") != plan_digest(plan):
+            return _failure(operation, operation_id, "plan_digest_mismatch")
+        public.update(
+            plan_id=response["plan_id"],
+            plan_digest=response["plan_digest"],
+            plan=plan,
+        )
+    return public
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Invoke a read-only configuration provider")
+    parser.add_argument("operation", choices=sorted(ALLOWED_OPERATIONS))
+    parser.add_argument("--operation-id", required=True)
+    parser.add_argument("--provider", nargs="+", required=True)
+    parser.add_argument("--enable", action="store_true")
+    parser.add_argument("--timeout", type=float, default=5)
+    args = parser.parse_args()
+    result = invoke_configuration_provider(
+        operation=args.operation,
+        operation_id=args.operation_id,
+        argv=args.provider,
+        enabled=args.enable,
+        timeout_seconds=args.timeout,
+    )
+    print(json.dumps(result, ensure_ascii=False))
+    return 0 if result["available"] else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_configuration_assembly_provider.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+from loopx.configuration_assembly_provider import (
+    SCHEMA_VERSION,
+    invoke_configuration_provider,
+    plan_digest,
+)
+
+
+def _provider(tmp_path: Path, response: dict | str, *, sleep: bool = False) -> list[str]:
+    body = "import sys,time; sys.stdin.read(); "
+    if sleep:
+        body += "time.sleep(1); "
+    body += f"print({json.dumps(json.dumps(response) if isinstance(response, dict) else response)})"
+    script = tmp_path / "provider.py"
+    script.write_text(body, encoding="utf-8")
+    return [sys.executable, str(script)]
+
+
+def test_provider_is_default_off_and_read_only() -> None:
+    result = invoke_configuration_provider(
+        operation="probe", operation_id="op-1", argv=["unused"]
+    )
+    assert result["failure_kind"] == "disabled"
+    assert result["status"] == "unknown"
+    assert result["available"] is False
+
+
+def test_plan_requires_bound_identity_and_digest(tmp_path: Path) -> None:
+    plan = {"revision": "public-v1", "state": "incomplete"}
+    response = {
+        "schema_version": SCHEMA_VERSION,
+        "operation": "plan",
+        "operation_id": "op-1",
+        "status": "incomplete",
+        "plan_id": "plan-1",
+        "plan_digest": plan_digest(plan),
+        "plan": plan,
+        "private": "must not escape",
+    }
+    result = invoke_configuration_provider(
+        operation="plan",
+        operation_id="op-1",
+        argv=_provider(tmp_path, response),
+        enabled=True,
+    )
+    assert result["status"] == "incomplete"
+    assert result["plan"] == plan
+    assert "private" not in result
+    assert "validated" not in result
+
+
+def test_rejects_schema_identity_and_digest_mismatch(tmp_path: Path) -> None:
+    base = {
+        "schema_version": SCHEMA_VERSION,
+        "operation": "plan",
+        "operation_id": "op-1",
+        "status": "ready",
+        "plan_id": "plan-1",
+        "plan_digest": "wrong",
+        "plan": {},
+    }
+    assert invoke_configuration_provider(
+        operation="plan", operation_id="op-1", argv=_provider(tmp_path, base), enabled=True
+    )["failure_kind"] == "plan_digest_mismatch"
+    base["schema_version"] = "future_v1"
+    assert invoke_configuration_provider(
+        operation="plan", operation_id="op-1", argv=_provider(tmp_path, base), enabled=True
+    )["failure_kind"] == "incompatible_schema"
+    base["schema_version"] = SCHEMA_VERSION
+    base["operation_id"] = "other"
+    assert invoke_configuration_provider(
+        operation="plan", operation_id="op-1", argv=_provider(tmp_path, base), enabled=True
+    )["failure_kind"] == "identity_mismatch"
+
+
+def test_provider_failures_are_public_safe(tmp_path: Path) -> None:
+    malformed = invoke_configuration_provider(
+        operation="probe", operation_id="op-1", argv=_provider(tmp_path, "{"), enabled=True
+    )
+    assert malformed["failure_kind"] == "malformed_response"
+    timeout = invoke_configuration_provider(
+        operation="probe",
+        operation_id="op-1",
+        argv=_provider(tmp_path, {}, sleep=True),
+        enabled=True,
+        timeout_seconds=0.01,
+    )
+    assert timeout["failure_kind"] == "timeout"
+    missing = invoke_configuration_provider(
+        operation="probe", operation_id="op-1", argv=[str(tmp_path / "missing")], enabled=True
+    )
+    assert missing["failure_kind"] == "missing_executable"