diff --git a/src/iac_code/cli/main.py b/src/iac_code/cli/main.py
index 4abfdf49..39592941 100644
--- a/src/iac_code/cli/main.py
+++ b/src/iac_code/cli/main.py
@@ -103,6 +103,7 @@ def main(
     model: str = typer.Option("", "--model", "-m", help=_("LLM model to use")),
     prompt: str = typer.Option("", "--prompt", "-p", help=_("Non-interactive mode: run a single prompt and exit")),
     output_format: str = typer.Option("text", "--output-format", help=_("Output format: text, json, stream-json")),
+    input_format: str = typer.Option("text", "--input-format", help=_("Input format: text, stream-json")),
     max_turns: int = typer.Option(100, "--max-turns", help=_("Maximum agent turns in headless mode")),
     debug: bool = typer.Option(False, "--debug", "-d", help=_("Enable debug logging")),
     verbose: bool = typer.Option(False, "--verbose", help=_("Show headless progress on stderr")),
@@ -177,6 +178,59 @@ def main(
         typer.echo(_("Error: --resume and --continue cannot be used together."), err=True)
         raise typer.Exit(1)
 
+    normalized_input_format = (input_format or "text").strip().lower()
+    if normalized_input_format not in {"text", "stream-json"}:
+        typer.echo(_("Invalid --input-format '{}'. Valid values: text, stream-json").format(input_format), err=True)
+        raise typer.Exit(1)
+
+    if normalized_input_format == "stream-json":
+        normalized_output_format = (output_format or "text").strip().lower()
+        if prompt:
+            typer.echo(_("--prompt cannot be used with --input-format stream-json."), err=True)
+            raise typer.Exit(1)
+        if normalized_output_format != "stream-json":
+            typer.echo(_("--input-format stream-json requires --output-format stream-json."), err=True)
+            raise typer.Exit(1)
+
+        from iac_code.pipeline.config import get_run_mode
+
+        run_mode = get_run_mode()
+
+        if not model:
+            try:
+                model = load_saved_model() or DEFAULT_MODEL
+            except ValueError as exc:
+                typer.echo(str(exc), err=True)
+                raise typer.Exit(1)
+
+        if permission_mode:
+            from iac_code.services.permissions.loader import parse_cli_permission_mode
+
+            try:
+                parse_cli_permission_mode(permission_mode)
+            except ValueError as exc:
+                typer.echo(str(exc), err=True)
+                raise typer.Exit(1) from exc
+
+        setup_logging(session_id="process-mode", debug=debug)
+
+        from iac_code.cli.process_mode import ProcessModeOptions, ProcessModeRunner
+
+        cli_allowed = [s.strip() for s in allowed_tools.split(",") if s.strip()] if allowed_tools else None
+        cli_disallowed = [s.strip() for s in disallowed_tools.split(",") if s.strip()] if disallowed_tools else None
+        runner = ProcessModeRunner(
+            ProcessModeOptions(
+                model=model,
+                cwd=os.getcwd(),
+                run_mode=run_mode.value,
+                max_turns=max_turns,
+                cli_allowed_tools=cli_allowed,
+                cli_disallowed_tools=cli_disallowed,
+                cli_permission_mode=permission_mode or None,
+            )
+        )
+        raise SystemExit(asyncio.run(runner.run()))
+
     fmt = None
     if prompt:
         from iac_code.cli.output_formats import OutputFormat
diff --git a/src/iac_code/cli/output_formats.py b/src/iac_code/cli/output_formats.py
index ce6306a9..3d6c7d12 100644
--- a/src/iac_code/cli/output_formats.py
+++ b/src/iac_code/cli/output_formats.py
@@ -81,7 +81,8 @@ def _sanitize_public_value(value: Any, *, public_path_roots: list[dict[str, str]
     return value
 
 
-def _stream_json_event_data(event: StreamEvent) -> dict[str, Any]:
+def stream_json_event_data(event: StreamEvent) -> dict[str, Any]:
+    """Return the public stream-json representation for a stream event."""
     if isinstance(event, ToolUseStartEvent):
         return {
             "tool_use_id": event.tool_use_id,
@@ -120,7 +121,7 @@ def _stream_json_event_data(event: StreamEvent) -> dict[str, Any]:
     if isinstance(event, SubPipelineStreamEvent):
         return {
             "candidate_index": event.candidate_index,
-            "inner": _stream_json_event_data(event.inner),
+            "inner": stream_json_event_data(event.inner),
             "sub_pipeline_id": event.sub_pipeline_id,
             "type": event.type,
         }
@@ -140,6 +141,10 @@ def _stream_json_event_data(event: StreamEvent) -> dict[str, Any]:
     return data
 
 
+def _stream_json_event_data(event: StreamEvent) -> dict[str, Any]:
+    return stream_json_event_data(event)
+
+
 class TextWriter:
     """Writes only assistant text content to the output stream.
 
@@ -231,7 +236,7 @@ def __init__(self, stream: IO[str] | None = None) -> None:
         self._stream = stream or sys.stdout
 
     def handle(self, event: StreamEvent) -> None:
-        data = _stream_json_event_data(event)
+        data = stream_json_event_data(event)
         self._stream.write(json.dumps(data, ensure_ascii=False, default=str))
         self._stream.write("\n")
         self._stream.flush()
diff --git a/src/iac_code/cli/process_events.py b/src/iac_code/cli/process_events.py
new file mode 100644
index 00000000..1231c929
--- /dev/null
+++ b/src/iac_code/cli/process_events.py
@@ -0,0 +1,56 @@
+"""Event and error mapping for CLI process mode."""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from typing import Any
+
+from iac_code.cli.output_formats import stream_json_event_data
+from iac_code.cli.process_protocol import ProcessFrameValidationError, SDKErrorPayload, SDKProcessRuntimeError
+from iac_code.providers.manager import ProviderNotConfiguredError
+from iac_code.types.stream_events import ErrorEvent
+from iac_code.utils.public_errors import sanitize_public_text
+
+
+@dataclass(frozen=True)
+class ProcessSerializedEvent:
+    """Already-public event payload to embed in a process-mode stream_event frame."""
+
+    payload: dict[str, Any]
+
+
+class ProcessEventSerializer:
+    """Serialize internal stream events to the public stream-json event shape."""
+
+    def serialize(self, event: Any) -> dict:
+        if isinstance(event, ProcessSerializedEvent):
+            return event.payload
+        return stream_json_event_data(event)
+
+
+class ProcessErrorMapper:
+    """Map runtime exceptions and error events to stable process-mode error payloads."""
+
+    def from_event(self, event: ErrorEvent) -> SDKErrorPayload:
+        return SDKErrorPayload(
+            code="stream_error",
+            message=sanitize_public_text(event.error),
+            retryable=event.is_retryable,
+            error_id=event.error_id,
+        )
+
+    def from_exception(self, exc: BaseException) -> SDKErrorPayload:
+        if isinstance(exc, SDKProcessRuntimeError):
+            return exc.payload
+        if isinstance(exc, ProcessFrameValidationError):
+            return SDKErrorPayload(code=exc.code, message=exc.message, retryable=exc.retryable)
+        if isinstance(exc, ProviderNotConfiguredError):
+            return SDKErrorPayload(
+                code="provider_not_configured",
+                message=sanitize_public_text(str(exc)),
+                retryable=False,
+            )
+        if isinstance(exc, asyncio.CancelledError):
+            return SDKErrorPayload(code="turn_canceled", message="Turn canceled.", retryable=False)
+        return SDKErrorPayload(code="internal_error", message=sanitize_public_text(str(exc)), retryable=False)
diff --git a/src/iac_code/cli/process_mode.py b/src/iac_code/cli/process_mode.py
new file mode 100644
index 00000000..73a1e51d
--- /dev/null
+++ b/src/iac_code/cli/process_mode.py
@@ -0,0 +1,1084 @@
+"""Long-running stream-json process mode for local SDK subprocess clients."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import sys
+import time
+import uuid
+from dataclasses import dataclass, field, replace
+from pathlib import Path
+from typing import IO, Any
+from urllib.parse import quote
+
+from loguru import logger
+
+from iac_code.cli.process_events import ProcessErrorMapper, ProcessEventSerializer, ProcessSerializedEvent
+from iac_code.cli.process_protocol import (
+    ProcessFrameParser,
+    ProcessFrameValidationError,
+    ProcessInputMessage,
+    SDKControlRequest,
+    SDKControlResponse,
+    SDKErrorPayload,
+    SDKProcessRuntimeError,
+    SDKUpdateEnvironmentVariables,
+    SDKUserMessage,
+)
+from iac_code.types.stream_events import (
+    ErrorEvent,
+    MessageEndEvent,
+    PermissionRequestEvent,
+    SubPipelineStreamEvent,
+    TextDeltaEvent,
+    Usage,
+)
+from iac_code.utils.project_paths import get_session_path
+
+EXIT_OK = 0
+EXIT_ERROR = 1
+
+
+@dataclass(frozen=True)
+class ProcessModeOptions:
+    model: str
+    cwd: str
+    run_mode: str = "normal"
+    max_turns: int = 100
+    cli_allowed_tools: list[str] | None = None
+    cli_disallowed_tools: list[str] | None = None
+    cli_permission_mode: str | None = None
+
+
+class ProcessTransport:
+    """Line-oriented stdin/stdout transport for process mode."""
+
+    def __init__(self, input_stream: IO[str] | None = None, output_stream: IO[str] | None = None) -> None:
+        self._input_stream = input_stream or sys.stdin
+        self._output_stream = output_stream or sys.stdout
+        self._write_lock = asyncio.Lock()
+
+    async def readline(self) -> str:
+        return await asyncio.to_thread(self._input_stream.readline)
+
+    async def write_frame(self, frame: dict[str, Any]) -> None:
+        line = json.dumps(frame, ensure_ascii=False, default=str)
+        async with self._write_lock:
+            self._output_stream.write(line)
+            self._output_stream.write("\n")
+            self._output_stream.flush()
+
+
+class ProcessSessionLock:
+    """Cross-process advisory lock for a single cwd/session_id pair."""
+
+    def __init__(self, *, cwd: str, session_id: str) -> None:
+        session_path = get_session_path(cwd, session_id)
+        self._lock_path = session_path.with_name(f".{session_path.name}.process.lock")
+        self._lock_file: IO[bytes] | None = None
+
+    def acquire(self, *, blocking: bool = False) -> bool:
+        if self._lock_file is not None:
+            return True
+        self._lock_path.parent.mkdir(parents=True, exist_ok=True)
+        lock_file = self._lock_path.open("a+b")
+        try:
+            if sys.platform == "win32":
+                import msvcrt
+
+                lock_file.seek(0)
+                mode = msvcrt.LK_LOCK if blocking else msvcrt.LK_NBLCK
+                msvcrt.locking(lock_file.fileno(), mode, 1)
+            else:
+                import fcntl
+
+                flags = fcntl.LOCK_EX if blocking else fcntl.LOCK_EX | fcntl.LOCK_NB
+                fcntl.flock(lock_file.fileno(), flags)
+        except OSError:
+            lock_file.close()
+            return False
+        self._lock_file = lock_file
+        return True
+
+    def release(self) -> None:
+        lock_file = self._lock_file
+        self._lock_file = None
+        if lock_file is None:
+            return
+        try:
+            if sys.platform == "win32":
+                import msvcrt
+
+                lock_file.seek(0)
+                msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
+            else:
+                import fcntl
+
+                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
+        finally:
+            lock_file.close()
+
+    def __enter__(self) -> "ProcessSessionLock":
+        if not self.acquire(blocking=False):
+            raise RuntimeError("session_busy")
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> None:
+        self.release()
+
+
+class ProcessRuntimeController:
+    """Create iac-code runtimes and execute user turns."""
+
+    def __init__(self, options: ProcessModeOptions) -> None:
+        self._options = options
+        self.model = options.model
+        self._cwd = options.cwd
+        self.session_id: str | None = None
+
+    async def initialize(self, frame: SDKControlRequest) -> dict[str, Any]:
+        model = frame.payload.get("model")
+        if isinstance(model, str) and model:
+            self.model = model
+        cwd = frame.payload.get("cwd")
+        if isinstance(cwd, str) and cwd:
+            self._cwd = cwd
+        return {
+            "protocol_version": "1.0",
+            "capabilities": [
+                "user",
+                "interrupt",
+                "set_model",
+                "end_session",
+                "close",
+                "keep_alive",
+                "update_environment_variables",
+            ],
+            "commands": [],
+            "agents": [],
+            "output_style": "default",
+            "available_output_styles": ["default"],
+            "models": [
+                {
+                    "value": self.model,
+                    "displayName": self.model,
+                    "description": "Current iac-code model",
+                }
+            ],
+            "account": {},
+            "cwd": self._cwd,
+            "pid": os.getpid(),
+        }
+
+    def set_model(self, model: str) -> None:
+        self.model = model
+
+    async def run_turn(self, frame: SDKUserMessage):
+        from iac_code.services.agent_factory import AgentFactoryOptions, create_agent_runtime
+
+        cwd = frame.cwd or self._cwd or os.getcwd()
+        session_lock = ProcessSessionLock(cwd=cwd, session_id=frame.session_id) if frame.session_id else None
+        if session_lock is not None and not session_lock.acquire(blocking=False):
+            raise SessionBusyError("session is busy")
+
+        runtime = None
+        try:
+            runtime = create_agent_runtime(
+                AgentFactoryOptions(
+                    model=self.model,
+                    session_id=frame.session_id,
+                    cwd=cwd,
+                    max_turns=self._options.max_turns,
+                    cli_allowed_tools=self._options.cli_allowed_tools,
+                    cli_disallowed_tools=self._options.cli_disallowed_tools,
+                    cli_permission_mode=self._options.cli_permission_mode,
+                )
+            )
+            self.session_id = runtime.session_id
+            async for event in runtime.agent_loop.run_streaming(frame.text):
+                permission_event = _permission_request_event(event)
+                if permission_event is not None:
+                    _auto_answer_permission(permission_event)
+                    continue
+                yield event
+        finally:
+            if session_lock is not None:
+                session_lock.release()
+            close = getattr(runtime, "aclose", None)
+            if close is not None:
+                try:
+                    await close()
+                except Exception:
+                    logger.debug("Process mode runtime close failed", exc_info=True)
+
+    async def aclose(self) -> None:
+        return None
+
+
+class SessionBusyError(RuntimeError):
+    """Raised when another process holds a session lock."""
+
+
+@dataclass
+class ProcessTurnHandle:
+    request_id: str
+    session_id: str | None
+    task: asyncio.Task
+
+    def cancel(self) -> None:
+        if not self.task.done():
+            self.task.cancel()
+
+
+@dataclass(frozen=True)
+class ProcessResultPatch:
+    """Controller-supplied updates for the final result frame."""
+
+    stop_reason: str | None = None
+    subtype: str | None = None
+    is_error: bool | None = None
+    result: str | None = None
+    errors: list[str] = field(default_factory=list)
+    extra_fields: dict[str, Any] = field(default_factory=dict)
+
+
+class ProcessTurnResult:
+    """Collect Claude-style result fields while a turn streams."""
+
+    def __init__(self) -> None:
+        self._text_chunks: list[str] = []
+        self._usage = Usage()
+        self._stop_reason: str | None = None
+        self._errors: list[str] = []
+        self._is_error = False
+        self._subtype = "success"
+        self._result_override: str | None = None
+        self._extra_fields: dict[str, Any] = {}
+
+    def observe(self, event: Any, error_mapper: ProcessErrorMapper) -> None:
+        if isinstance(event, ProcessResultPatch):
+            if event.stop_reason is not None:
+                self._stop_reason = event.stop_reason
+            if event.subtype is not None:
+                self._subtype = event.subtype
+            if event.is_error is not None:
+                self._is_error = event.is_error
+            if event.result is not None:
+                self._result_override = event.result
+            self._errors.extend(event.errors)
+            self._extra_fields.update(event.extra_fields)
+            return
+        if isinstance(event, TextDeltaEvent):
+            self._text_chunks.append(event.text)
+            return
+        if isinstance(event, MessageEndEvent):
+            self._usage = event.usage
+            self._stop_reason = event.stop_reason
+            return
+        if isinstance(event, ErrorEvent):
+            payload = error_mapper.from_event(event)
+            self.mark_error(payload.message)
+
+    def mark_error(self, message: str, *, stop_reason: str | None = None) -> None:
+        self._is_error = True
+        self._subtype = "error_during_execution"
+        self._errors.append(message)
+        if stop_reason is not None:
+            self._stop_reason = stop_reason
+
+    def as_frame(self, *, request_id: str, session_id: str, duration_ms: int) -> dict[str, Any]:
+        base: dict[str, Any] = {
+            "type": "result",
+            "request_id": request_id,
+            "subtype": self._subtype,
+            "duration_ms": duration_ms,
+            "duration_api_ms": 0,
+            "is_error": self._is_error,
+            "num_turns": 1,
+            "stop_reason": self._stop_reason,
+            "total_cost_usd": 0.0,
+            "usage": {
+                "input_tokens": self._usage.input_tokens,
+                "output_tokens": self._usage.output_tokens,
+                "cache_creation_input_tokens": self._usage.cache_creation_input_tokens,
+                "cache_read_input_tokens": self._usage.cache_read_input_tokens,
+            },
+            "modelUsage": {},
+            "permission_denials": [],
+            "uuid": str(uuid.uuid4()),
+            "session_id": session_id,
+        }
+        if self._is_error:
+            base["errors"] = self._errors
+        else:
+            base["result"] = self._result_override if self._result_override is not None else "".join(self._text_chunks)
+        base.update(self._extra_fields)
+        return base
+
+
+@dataclass(frozen=True)
+class PipelineProcessCreateRequest:
+    context_id: str
+    task_id: str
+    iac_code_session_id: str
+    cwd: str
+    model: str
+    resume_from_sidecar: bool
+    agent_runtime: Any | None = None
+
+
+@dataclass(frozen=True)
+class PipelineProcessContextSnapshot:
+    context_id: str
+    task_id: str
+    iac_code_session_id: str
+    cwd: str
+    sidecar_status: str | None = None
+    active_task_id: str | None = None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "PipelineProcessContextSnapshot":
+        return cls(
+            context_id=str(data["contextId"]),
+            task_id=str(data["taskId"]),
+            iac_code_session_id=str(data["iacCodeSessionId"]),
+            cwd=str(data["cwd"]),
+            sidecar_status=data.get("sidecarStatus") if isinstance(data.get("sidecarStatus"), str) else None,
+            active_task_id=data.get("activeTaskId") if isinstance(data.get("activeTaskId"), str) else None,
+        )
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "contextId": self.context_id,
+            "taskId": self.task_id,
+            "iacCodeSessionId": self.iac_code_session_id,
+            "cwd": self.cwd,
+            "sidecarStatus": self.sidecar_status,
+            "activeTaskId": self.active_task_id,
+        }
+
+
+class PipelineProcessContextLock:
+    """Cross-process advisory lock for one pipeline context."""
+
+    def __init__(self, lock_path: Path) -> None:
+        self._lock_path = lock_path
+        self._lock_file: IO[bytes] | None = None
+
+    def acquire(self, *, blocking: bool = True) -> bool:
+        if self._lock_file is not None:
+            return True
+        self._lock_path.parent.mkdir(parents=True, exist_ok=True)
+        lock_file = self._lock_path.open("a+b")
+        try:
+            if sys.platform == "win32":
+                import msvcrt
+
+                lock_file.seek(0)
+                mode = msvcrt.LK_LOCK if blocking else msvcrt.LK_NBLCK
+                msvcrt.locking(lock_file.fileno(), mode, 1)
+            else:
+                import fcntl
+
+                flags = fcntl.LOCK_EX if blocking else fcntl.LOCK_EX | fcntl.LOCK_NB
+                fcntl.flock(lock_file.fileno(), flags)
+        except OSError:
+            lock_file.close()
+            return False
+        self._lock_file = lock_file
+        return True
+
+    def release(self) -> None:
+        lock_file = self._lock_file
+        self._lock_file = None
+        if lock_file is None:
+            return
+        try:
+            if sys.platform == "win32":
+                import msvcrt
+
+                lock_file.seek(0)
+                msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
+            else:
+                import fcntl
+
+                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
+        finally:
+            lock_file.close()
+
+    def __enter__(self) -> "PipelineProcessContextLock":
+        self.acquire(blocking=True)
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> None:
+        self.release()
+
+
+class PipelineProcessContextStore:
+    """Small process-mode store for pipeline context/task recovery."""
+
+    def __init__(self, root: Path | None = None) -> None:
+        self._root = root
+
+    @property
+    def root(self) -> Path:
+        if self._root is not None:
+            return self._root
+        from iac_code.config import get_config_dir
+
+        return get_config_dir() / "process-pipeline" / "contexts"
+
+    def lock(self, context_id: str) -> PipelineProcessContextLock:
+        return PipelineProcessContextLock(self._path_for(context_id).with_suffix(".lock"))
+
+    def load(self, context_id: str) -> PipelineProcessContextSnapshot | None:
+        path = self._path_for(context_id)
+        if not path.exists():
+            return None
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            logger.exception("Failed to load process pipeline context %s", context_id)
+            return None
+        if not isinstance(data, dict):
+            return None
+        try:
+            return PipelineProcessContextSnapshot.from_dict(data)
+        except (KeyError, TypeError, ValueError):
+            logger.exception("Invalid process pipeline context snapshot %s", context_id)
+            return None
+
+    def save(self, snapshot: PipelineProcessContextSnapshot) -> None:
+        path = self._path_for(snapshot.context_id)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = path.with_suffix(".tmp")
+        tmp.write_text(json.dumps(snapshot.as_dict(), ensure_ascii=False, indent=2), encoding="utf-8")
+        tmp.replace(path)
+
+    def _path_for(self, context_id: str) -> Path:
+        return self.root / f"{quote(context_id, safe='')}.json"
+
+
+@dataclass
+class PipelineProcessTurnState:
+    status: str = "completed"
+    sidecar_status: str | None = None
+    stop_reason: str = "end_turn"
+    is_error: bool = False
+
+    def observe(self, event: Any) -> None:
+        from iac_code.pipeline.engine.events import PipelineEvent, PipelineEventType
+
+        if not isinstance(event, PipelineEvent):
+            return
+        if event.type == PipelineEventType.USER_INPUT_REQUIRED:
+            self.status = "input_required"
+            self.sidecar_status = "waiting_input"
+            self.stop_reason = "input_required"
+            return
+        if event.type == PipelineEventType.BACKUP_BLOCKED:
+            self.status = "input_required"
+            self.sidecar_status = "backup_blocked"
+            self.stop_reason = "input_required"
+            return
+        if event.type == PipelineEventType.PIPELINE_COMPLETED:
+            failed = event.data.get("failed") is True if isinstance(event.data, dict) else False
+            self.status = "failed" if failed else "completed"
+            self.sidecar_status = self.status
+            self.stop_reason = "error" if failed else "end_turn"
+            self.is_error = failed
+            return
+        if event.type == PipelineEventType.PIPELINE_ERROR:
+            self.status = "failed"
+            self.sidecar_status = "failed"
+            self.stop_reason = "error"
+            self.is_error = True
+            return
+        if event.type == PipelineEventType.INTERRUPTED:
+            self.status = "canceled"
+            self.sidecar_status = "canceled"
+            self.stop_reason = "cancelled"
+            self.is_error = True
+
+
+class PipelineProcessRuntimeController:
+    """Execute pipeline mode through the SDK process protocol."""
+
+    def __init__(
+        self,
+        options: ProcessModeOptions,
+        *,
+        agent_runtime_factory: Any | None = None,
+        pipeline_factory: Any | None = None,
+        context_store: PipelineProcessContextStore | None = None,
+    ) -> None:
+        self._options = options
+        self.model = options.model
+        self._cwd = options.cwd
+        self._agent_runtime_factory = agent_runtime_factory or self._default_agent_runtime_factory
+        self._pipeline_factory = pipeline_factory or self._default_pipeline_factory
+        self._context_store = context_store or PipelineProcessContextStore()
+        self._agent_runtime: Any | None = None
+        self._pipeline: Any | None = None
+        self._translator: Any | None = None
+        self._context_id: str | None = None
+        self._task_id: str | None = None
+        self.session_id: str | None = None
+
+    async def initialize(self, frame: SDKControlRequest) -> dict[str, Any]:
+        model = frame.payload.get("model")
+        if isinstance(model, str) and model:
+            self.model = model
+        cwd = frame.payload.get("cwd")
+        if isinstance(cwd, str) and cwd:
+            self._cwd = cwd
+        return {
+            "protocol_version": "1.0",
+            "capabilities": [
+                "user",
+                "interrupt",
+                "set_model",
+                "end_session",
+                "close",
+                "keep_alive",
+                "update_environment_variables",
+                "pipeline",
+                "pipeline_resume",
+                "pipeline_recoverable_task",
+                "pipeline_cancel",
+            ],
+            "commands": [],
+            "agents": [],
+            "output_style": "default",
+            "available_output_styles": ["default"],
+            "models": [
+                {
+                    "value": self.model,
+                    "displayName": self.model,
+                    "description": "Current iac-code model",
+                }
+            ],
+            "account": {},
+            "cwd": self._cwd,
+            "pid": os.getpid(),
+        }
+
+    def set_model(self, model: str) -> None:
+        self.model = model
+
+    async def run_turn(self, frame: SDKUserMessage):
+        from iac_code.pipeline.engine.user_input import normalize_pipeline_user_input
+
+        cwd = frame.cwd or self._cwd or os.getcwd()
+        metadata = _pipeline_metadata(frame.metadata)
+        requested_context_id = _first_metadata_string(metadata, "contextId", "context_id")
+        requested_task_id = _first_metadata_string(metadata, "taskId", "task_id")
+        requested_iac_session_id = _first_metadata_string(metadata, "iacCodeSessionId", "iac_code_session_id")
+        context_id = requested_context_id or self._context_id or str(uuid.uuid4())
+
+        with self._context_store.lock(context_id):
+            snapshot = self._context_store.load(context_id)
+            self._validate_pipeline_context(snapshot, cwd=cwd, task_id=requested_task_id)
+            task_id = requested_task_id or (snapshot.task_id if snapshot and snapshot.active_task_id is None else None)
+            if task_id is None:
+                if snapshot is not None and snapshot.active_task_id:
+                    self._raise_recoverable_task(snapshot)
+                task_id = str(uuid.uuid4())
+            iac_code_session_id = (
+                requested_iac_session_id
+                or (snapshot.iac_code_session_id if snapshot is not None else None)
+                or frame.session_id
+                or str(uuid.uuid4())[:8]
+            )
+            running_snapshot = PipelineProcessContextSnapshot(
+                context_id=context_id,
+                task_id=task_id,
+                iac_code_session_id=iac_code_session_id,
+                cwd=cwd,
+                sidecar_status="running",
+                active_task_id=task_id,
+            )
+            self._context_store.save(running_snapshot)
+
+        request = PipelineProcessCreateRequest(
+            context_id=context_id,
+            task_id=task_id,
+            iac_code_session_id=iac_code_session_id,
+            cwd=cwd,
+            model=self.model,
+            resume_from_sidecar=snapshot is not None,
+        )
+        pipeline = self._ensure_pipeline(request)
+        pipeline_input = normalize_pipeline_user_input(frame.text)
+        stream = self._pipeline_stream(pipeline, pipeline_input, snapshot)
+        state = PipelineProcessTurnState()
+
+        async for event in stream:
+            state.observe(event)
+            for payload in self._translate_pipeline_event(event, request):
+                yield ProcessSerializedEvent({"type": "pipeline_event", **payload})
+            permission_event = _permission_request_event(event)
+            if permission_event is not None:
+                _auto_answer_permission(permission_event)
+
+        sidecar_status = getattr(pipeline, "sidecar_status", None) or state.sidecar_status or state.status
+        state.sidecar_status = sidecar_status
+        if sidecar_status in {"waiting_input", "backup_blocked"}:
+            state.status = "input_required"
+            state.stop_reason = "input_required"
+        await self._persist_pipeline_result(request, state)
+        self._context_id = context_id
+        self._task_id = task_id
+        self.session_id = iac_code_session_id
+        yield ProcessResultPatch(
+            stop_reason=state.stop_reason,
+            subtype="error_during_execution" if state.is_error else "success",
+            is_error=state.is_error,
+            result="",
+            extra_fields={
+                "pipeline": {
+                    "mode": "pipeline",
+                    "name": self._pipeline_name(),
+                    "contextId": context_id,
+                    "taskId": task_id,
+                    "iacCodeSessionId": iac_code_session_id,
+                    "status": state.status,
+                    "sidecarStatus": state.sidecar_status,
+                }
+            },
+        )
+
+    async def aclose(self) -> None:
+        close = getattr(self._agent_runtime, "aclose", None)
+        if close is None:
+            return
+        try:
+            await close()
+        except Exception:
+            logger.debug("Pipeline process runtime close failed", exc_info=True)
+
+    def _validate_pipeline_context(
+        self, snapshot: PipelineProcessContextSnapshot | None, *, cwd: str, task_id: str | None
+    ) -> None:
+        if snapshot is None:
+            return
+        if snapshot.cwd != cwd:
+            raise SDKProcessRuntimeError(
+                SDKErrorPayload(
+                    code="pipeline_context_mismatch",
+                    message="Pipeline context belongs to a different cwd.",
+                    retryable=False,
+                    data={"contextId": snapshot.context_id, "cwd": snapshot.cwd},
+                )
+            )
+        if snapshot.active_task_id is not None and task_id != snapshot.active_task_id:
+            self._raise_recoverable_task(snapshot)
+
+    def _raise_recoverable_task(self, snapshot: PipelineProcessContextSnapshot) -> None:
+        raise SDKProcessRuntimeError(
+            SDKErrorPayload(
+                code="pipeline_task_required",
+                message="Pipeline context already has a recoverable task.",
+                retryable=True,
+                data={
+                    "contextId": snapshot.context_id,
+                    "recoverableTaskId": snapshot.active_task_id or snapshot.task_id,
+                    "sidecarStatus": snapshot.sidecar_status,
+                },
+            )
+        )
+
+    def _ensure_pipeline(self, request: PipelineProcessCreateRequest) -> Any:
+        if self._pipeline is not None and self._context_id == request.context_id and self._task_id == request.task_id:
+            return self._pipeline
+        self._agent_runtime = self._agent_runtime_factory(request)
+        request = replace(request, agent_runtime=self._agent_runtime)
+        self._pipeline = self._pipeline_factory(request)
+        self._translator = self._create_translator(request)
+        self._context_id = request.context_id
+        self._task_id = request.task_id
+        self.session_id = request.iac_code_session_id
+        return self._pipeline
+
+    def _pipeline_stream(self, pipeline: Any, pipeline_input: Any, snapshot: PipelineProcessContextSnapshot | None):
+        sidecar_status = getattr(pipeline, "sidecar_status", None) or (snapshot.sidecar_status if snapshot else None)
+        if sidecar_status in {"waiting_input", "backup_blocked"}:
+            if sidecar_status == "backup_blocked" and hasattr(pipeline, "continue_from_sidecar"):
+                return pipeline.continue_from_sidecar(pipeline_input)
+            return pipeline.resume(pipeline_input)
+        return pipeline.run(pipeline_input)
+
+    def _translate_pipeline_event(self, event: Any, request: PipelineProcessCreateRequest) -> list[dict[str, Any]]:
+        if self._translator is None:
+            self._translator = self._create_translator(request)
+        return self._translator.translate(event)
+
+    def _create_translator(self, request: PipelineProcessCreateRequest) -> Any:
+        from iac_code.a2a.pipeline_events import PipelineA2AContext, PipelineEventTranslator
+
+        return PipelineEventTranslator(
+            PipelineA2AContext(
+                pipeline_run_id=request.context_id,
+                task_id=request.task_id,
+                context_id=request.context_id,
+                pipeline_name=self._pipeline_name(),
+                iac_code_session_id=request.iac_code_session_id,
+            )
+        )
+
+    async def _persist_pipeline_result(
+        self, request: PipelineProcessCreateRequest, state: PipelineProcessTurnState
+    ) -> None:
+        active_task_id = request.task_id if state.status in {"input_required", "working"} else None
+        with self._context_store.lock(request.context_id):
+            self._context_store.save(
+                PipelineProcessContextSnapshot(
+                    context_id=request.context_id,
+                    task_id=request.task_id,
+                    iac_code_session_id=request.iac_code_session_id,
+                    cwd=request.cwd,
+                    sidecar_status=state.sidecar_status,
+                    active_task_id=active_task_id,
+                )
+            )
+
+    def _default_agent_runtime_factory(self, request: PipelineProcessCreateRequest) -> Any:
+        from iac_code.services.agent_factory import AgentFactoryOptions, create_agent_runtime
+
+        return create_agent_runtime(
+            AgentFactoryOptions(
+                model=request.model,
+                session_id=request.iac_code_session_id,
+                cwd=request.cwd,
+                max_turns=self._options.max_turns,
+                cli_allowed_tools=self._options.cli_allowed_tools,
+                cli_disallowed_tools=self._options.cli_disallowed_tools,
+                cli_permission_mode=self._options.cli_permission_mode,
+            )
+        )
+
+    def _default_pipeline_factory(self, request: PipelineProcessCreateRequest) -> Any:
+        from iac_code.pipeline import create_pipeline
+        from iac_code.services.session_backup import SessionBackupService
+        from iac_code.services.session_storage import SessionStorage
+
+        runtime = request.agent_runtime
+        if runtime is None:
+            raise RuntimeError("pipeline process requires an agent runtime")
+        agent_loop = getattr(runtime, "agent_loop", None)
+        session_storage = getattr(agent_loop, "_session_storage", None)
+        if session_storage is None:
+            session_storage = SessionStorage()
+        command_registry = getattr(runtime, "command_registry", None)
+        skills = command_registry.get_model_invocable_skills() if command_registry is not None else None
+        permission_context = getattr(agent_loop, "permission_context", None)
+        return create_pipeline(
+            self._pipeline_name(),
+            provider_manager=runtime.provider_manager,
+            base_tool_registry=runtime.tool_registry,
+            session_storage=session_storage,
+            session_id=request.iac_code_session_id,
+            cwd=request.cwd,
+            permission_context_getter=lambda: permission_context,
+            auto_trigger_skills=skills,
+            resume_from_sidecar=request.resume_from_sidecar,
+            surface="process",
+            backup_service=SessionBackupService(session_storage=session_storage),
+        )
+
+    def _pipeline_name(self) -> str:
+        from iac_code.pipeline.config import get_pipeline_name
+
+        return get_pipeline_name()
+
+
+class ProcessRuntimeControllerFactory:
+    """Select the process runtime controller for the configured run mode."""
+
+    @staticmethod
+    def create(options: ProcessModeOptions) -> Any:
+        if options.run_mode == "pipeline":
+            return PipelineProcessRuntimeController(options)
+        return ProcessRuntimeController(options)
+
+
+class ProcessQuery:
+    """Bidirectional control protocol manager for process mode."""
+
+    def __init__(
+        self,
+        *,
+        transport: ProcessTransport,
+        runtime_controller: Any,
+        parser: ProcessFrameParser | None = None,
+        serializer: ProcessEventSerializer | None = None,
+        error_mapper: ProcessErrorMapper | None = None,
+    ) -> None:
+        self._transport = transport
+        self._runtime_controller = runtime_controller
+        self._parser = parser or ProcessFrameParser()
+        self._serializer = serializer or ProcessEventSerializer()
+        self._error_mapper = error_mapper or ProcessErrorMapper()
+        self._initialized = False
+        self._active_turn: ProcessTurnHandle | None = None
+        self._turn_counter = 0
+        self._exit_code = EXIT_OK
+        self._stop_requested = False
+
+    async def run(self) -> int:
+        try:
+            while not self._stop_requested:
+                line = await self._transport.readline()
+                if line == "":
+                    break
+                try:
+                    frame = self._parser.parse_line(line)
+                except ProcessFrameValidationError as exc:
+                    await self._write_error(self._error_mapper.from_exception(exc), exc.request_id)
+                    self._exit_code = EXIT_ERROR
+                    continue
+                if frame is None:
+                    continue
+                await self._handle_frame(frame)
+            if self._active_turn is not None:
+                await self._wait_active_turn()
+        finally:
+            close = getattr(self._runtime_controller, "aclose", None)
+            if close is not None:
+                await close()
+        return self._exit_code
+
+    async def _handle_frame(self, frame: ProcessInputMessage) -> None:
+        if isinstance(frame, SDKControlRequest):
+            await self._handle_control(frame)
+            return
+        if isinstance(frame, SDKControlResponse):
+            return
+        if isinstance(frame, SDKUpdateEnvironmentVariables):
+            os.environ.update(frame.variables)
+            return
+        await self._handle_user(frame)
+
+    async def _handle_control(self, frame: SDKControlRequest) -> None:
+        subtype = frame.subtype
+        if subtype == "initialize":
+            if self._initialized:
+                await self._write_control_error(
+                    frame.request_id,
+                    "already_initialized",
+                    "process mode is already initialized",
+                )
+                self._exit_code = EXIT_ERROR
+                return
+            response = await self._runtime_controller.initialize(frame)
+            self._initialized = True
+            await self._write_control_success(frame.request_id, response)
+            return
+        if subtype in {"close", "end_session"}:
+            if self._active_turn is not None:
+                self._active_turn.cancel()
+                await self._wait_active_turn()
+            await self._write_control_success(frame.request_id, {"ok": True})
+            self._stop_requested = True
+            return
+        if not self._initialized:
+            await self._write_control_error(
+                frame.request_id,
+                "not_initialized",
+                "initialize is required before control requests",
+            )
+            self._exit_code = EXIT_ERROR
+            return
+        if subtype == "interrupt":
+            if self._active_turn is not None:
+                self._active_turn.cancel()
+            await self._write_control_success(frame.request_id, {"ok": True})
+            return
+        if subtype == "set_model":
+            model = frame.payload.get("model")
+            if not isinstance(model, str) or not model:
+                await self._write_control_error(frame.request_id, "invalid_frame", "set_model requires model")
+                self._exit_code = EXIT_ERROR
+                return
+            self._runtime_controller.set_model(model)
+            await self._write_control_success(frame.request_id, {"ok": True})
+            return
+        await self._write_control_error(
+            frame.request_id,
+            "unsupported_control",
+            f"Unsupported control subtype: {subtype}",
+        )
+        self._exit_code = EXIT_ERROR
+
+    async def _handle_user(self, frame: SDKUserMessage) -> None:
+        if not self._initialized:
+            await self._write_error(
+                SDKErrorPayload(
+                    code="not_initialized",
+                    message="initialize is required before user messages",
+                    retryable=False,
+                ),
+                frame.request_id,
+            )
+            self._exit_code = EXIT_ERROR
+            return
+        if self._active_turn is not None and not self._active_turn.task.done():
+            await self._write_error(
+                SDKErrorPayload(code="turn_active", message="another turn is already active", retryable=True),
+                frame.request_id,
+            )
+            return
+        request_id = frame.request_id or self._next_turn_request_id()
+        task = asyncio.create_task(self._run_turn(frame, request_id))
+        self._active_turn = ProcessTurnHandle(request_id=request_id, session_id=frame.session_id, task=task)
+
+    async def _run_turn(self, frame: SDKUserMessage, request_id: str) -> None:
+        started = time.monotonic()
+        result = ProcessTurnResult()
+        try:
+            async for event in self._runtime_controller.run_turn(frame):
+                result.observe(event, self._error_mapper)
+                if isinstance(event, ProcessResultPatch):
+                    continue
+                await self._write_stream_event(request_id, frame.session_id, event)
+        except asyncio.CancelledError as exc:
+            result.mark_error(self._error_mapper.from_exception(exc).message, stop_reason="cancelled")
+        except SessionBusyError:
+            result.mark_error("session is busy")
+            await self._write_error(
+                SDKErrorPayload(code="session_busy", message="session is busy", retryable=True),
+                request_id,
+            )
+        except SDKProcessRuntimeError as exc:
+            payload = self._error_mapper.from_exception(exc)
+            result.mark_error(payload.message)
+            await self._write_error(payload, request_id)
+        except Exception as exc:
+            result.mark_error(self._error_mapper.from_exception(exc).message)
+        finally:
+            session_id = frame.session_id or getattr(self._runtime_controller, "session_id", None) or ""
+            await self._transport.write_frame(
+                result.as_frame(
+                    request_id=request_id,
+                    session_id=session_id,
+                    duration_ms=int((time.monotonic() - started) * 1000),
+                )
+            )
+            if self._active_turn is not None and self._active_turn.request_id == request_id:
+                self._active_turn = None
+
+    async def _wait_active_turn(self) -> None:
+        handle = self._active_turn
+        if handle is None:
+            return
+        try:
+            await handle.task
+        except asyncio.CancelledError:
+            pass
+
+    def _next_turn_request_id(self) -> str:
+        self._turn_counter += 1
+        return f"turn-{self._turn_counter}"
+
+    async def _write_stream_event(self, request_id: str, session_id: str | None, event: Any) -> None:
+        await self._transport.write_frame(
+            {
+                "type": "stream_event",
+                "request_id": request_id,
+                "session_id": session_id or getattr(self._runtime_controller, "session_id", None) or "",
+                "parent_tool_use_id": None,
+                "uuid": str(uuid.uuid4()),
+                "event": self._serializer.serialize(event),
+            }
+        )
+
+    async def _write_error(self, payload: SDKErrorPayload, request_id: str | None) -> None:
+        await self._transport.write_frame({"type": "error", "request_id": request_id, "error": payload.as_dict()})
+
+    async def _write_control_success(self, request_id: str, response: dict[str, Any]) -> None:
+        await self._transport.write_frame(
+            {
+                "type": "control_response",
+                "response": {
+                    "subtype": "success",
+                    "request_id": request_id,
+                    "response": response,
+                },
+            }
+        )
+
+    async def _write_control_error(self, request_id: str, code: str, message: str) -> None:
+        await self._transport.write_frame(
+            {
+                "type": "control_response",
+                "response": {
+                    "subtype": "error",
+                    "request_id": request_id,
+                    "code": code,
+                    "error": message,
+                },
+            }
+        )
+
+
+class ProcessModeRunner:
+    """Top-level CLI runner for stream-json process mode."""
+
+    def __init__(
+        self,
+        options: ProcessModeOptions,
+        *,
+        input_stream: IO[str] | None = None,
+        output_stream: IO[str] | None = None,
+        runtime_controller: Any | None = None,
+    ) -> None:
+        self._options = options
+        self._transport = ProcessTransport(input_stream=input_stream, output_stream=output_stream)
+        self._runtime_controller = runtime_controller or ProcessRuntimeControllerFactory.create(options)
+
+    async def run(self) -> int:
+        query = ProcessQuery(transport=self._transport, runtime_controller=self._runtime_controller)
+        return await query.run()
+
+
+def _pipeline_metadata(metadata: dict[str, Any]) -> dict[str, Any]:
+    iac_code = metadata.get("iac_code")
+    return iac_code if isinstance(iac_code, dict) else {}
+
+
+def _first_metadata_string(metadata: dict[str, Any], *names: str) -> str | None:
+    for name in names:
+        value = metadata.get(name)
+        if isinstance(value, str) and value:
+            return value
+    return None
+
+
+def _permission_request_event(event: Any) -> PermissionRequestEvent | None:
+    if isinstance(event, PermissionRequestEvent):
+        return event
+    if isinstance(event, SubPipelineStreamEvent):
+        return _permission_request_event(event.inner)
+    return None
+
+
+def _auto_answer_permission(event: PermissionRequestEvent) -> None:
+    if event.response_future is None or event.response_future.done():
+        return
+    from iac_code.services.permissions.audit import (
+        emit_auto_permission_audit,
+        is_aliyun_api_non_read_only_permission_event,
+    )
+
+    approved = not is_aliyun_api_non_read_only_permission_event(event)
+    audit_ok = emit_auto_permission_audit(
+        event,
+        decision="allow" if approved else "deny",
+        scope="auto_approve" if approved else "auto_deny",
+        source="process_mode_auto_approve" if approved else "process_mode_auto_deny",
+    )
+    if approved and not audit_ok:
+        approved = False
+    event.response_future.set_result(approved)
diff --git a/src/iac_code/cli/process_protocol.py b/src/iac_code/cli/process_protocol.py
new file mode 100644
index 00000000..bb38a269
--- /dev/null
+++ b/src/iac_code/cli/process_protocol.py
@@ -0,0 +1,282 @@
+"""Protocol models for the CLI stream-json process mode."""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class ProcessFrameValidationError(ValueError):
+    code: str
+    message: str
+    request_id: str | None = None
+    retryable: bool = False
+
+    def __str__(self) -> str:
+        return self.message
+
+
+@dataclass(frozen=True)
+class SDKErrorPayload:
+    code: str
+    message: str
+    retryable: bool = False
+    error_id: str | None = None
+    data: dict[str, Any] | None = None
+
+    def as_dict(self) -> dict[str, Any]:
+        data: dict[str, Any] = {
+            "code": self.code,
+            "message": self.message,
+            "retryable": self.retryable,
+        }
+        if self.error_id:
+            data["error_id"] = self.error_id
+        if self.data is not None:
+            data["data"] = self.data
+        return data
+
+
+class SDKProcessRuntimeError(RuntimeError):
+    """Runtime error that should be returned to SDK clients as a public error frame."""
+
+    def __init__(self, payload: SDKErrorPayload) -> None:
+        super().__init__(payload.message)
+        self.payload = payload
+
+
+@dataclass(frozen=True)
+class SDKControlRequest:
+    request_id: str
+    subtype: str
+    payload: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class SDKControlResponse:
+    request_id: str
+    subtype: str
+    payload: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class SDKUserMessage:
+    request_id: str | None
+    session_id: str | None
+    text: str
+    metadata: dict[str, Any] = field(default_factory=dict)
+    cwd: str | None = None
+    parent_tool_use_id: str | None = None
+    uuid: str | None = None
+    timestamp: str | None = None
+
+
+@dataclass(frozen=True)
+class SDKUpdateEnvironmentVariables:
+    variables: dict[str, str]
+
+
+ProcessInputMessage = SDKControlRequest | SDKControlResponse | SDKUpdateEnvironmentVariables | SDKUserMessage
+
+
+class ProcessFrameParser:
+    """Parse stdin JSON lines into process-mode protocol objects."""
+
+    def parse_line(self, line: str) -> ProcessInputMessage | None:
+        stripped = line.strip()
+        if not stripped:
+            return None
+        try:
+            raw = json.loads(stripped)
+        except json.JSONDecodeError as exc:
+            raise ProcessFrameValidationError("invalid_json", "Invalid JSON frame.") from exc
+        if not isinstance(raw, dict):
+            raise ProcessFrameValidationError("invalid_frame", "Frame must be a JSON object.")
+
+        frame_type = raw.get("type")
+        request_id = _optional_string(raw.get("request_id")) or _optional_string(raw.get("id"))
+        if frame_type == "control_request":
+            return self._parse_control_request(raw, request_id)
+        if frame_type == "control_response":
+            return self._parse_control_response(raw, request_id)
+        if frame_type == "keep_alive":
+            return None
+        if frame_type == "update_environment_variables":
+            return self._parse_update_environment_variables(raw, request_id)
+        if frame_type == "control":
+            return self._parse_legacy_control(raw, request_id)
+        if frame_type == "initialize":
+            return self._parse_legacy_initialize(raw, request_id)
+        if frame_type == "close":
+            return self._parse_legacy_close(raw, request_id)
+        if frame_type == "user":
+            return self._parse_user(raw, request_id)
+        if frame_type == "user_message":
+            return self._parse_legacy_user(raw, request_id)
+        raise ProcessFrameValidationError("invalid_frame", f"Unsupported frame type: {frame_type!r}.", request_id)
+
+    def _parse_control_request(self, raw: dict[str, Any], request_id: str | None) -> SDKControlRequest:
+        if not request_id:
+            raise ProcessFrameValidationError("invalid_frame", "control_request requires request_id.")
+        request = raw.get("request")
+        if not isinstance(request, dict):
+            raise ProcessFrameValidationError("invalid_frame", "control_request.request must be an object.", request_id)
+        subtype = _required_string(request.get("subtype"), "control_request.request.subtype", request_id)
+        payload = dict(request)
+        self._validate_payload_paths(payload, request_id)
+        return SDKControlRequest(request_id=request_id, subtype=subtype, payload=payload)
+
+    def _parse_control_response(self, raw: dict[str, Any], request_id: str | None) -> SDKControlResponse:
+        response = raw.get("response")
+        if not isinstance(response, dict):
+            raise ProcessFrameValidationError(
+                "invalid_frame",
+                "control_response.response must be an object.",
+                request_id,
+            )
+        response_request_id = _optional_string(response.get("request_id")) or request_id
+        if not response_request_id:
+            raise ProcessFrameValidationError(
+                "invalid_frame", "control_response.response.request_id is required.", request_id
+            )
+        subtype = _required_string(response.get("subtype"), "control_response.response.subtype", response_request_id)
+        return SDKControlResponse(request_id=response_request_id, subtype=subtype, payload=dict(response))
+
+    def _parse_update_environment_variables(
+        self, raw: dict[str, Any], request_id: str | None
+    ) -> SDKUpdateEnvironmentVariables:
+        variables = raw.get("variables")
+        if not isinstance(variables, dict):
+            raise ProcessFrameValidationError(
+                "invalid_frame", "update_environment_variables.variables must be an object.", request_id
+            )
+        parsed: dict[str, str] = {}
+        for key, value in variables.items():
+            if not isinstance(key, str) or not isinstance(value, str):
+                raise ProcessFrameValidationError(
+                    "invalid_frame", "update_environment_variables variables must be string pairs.", request_id
+                )
+            parsed[key] = value
+        return SDKUpdateEnvironmentVariables(variables=parsed)
+
+    def _parse_legacy_control(self, raw: dict[str, Any], request_id: str | None) -> SDKControlRequest:
+        if not request_id:
+            raise ProcessFrameValidationError("invalid_frame", "control frame requires id.")
+        subtype = _required_string(raw.get("subtype"), "control.subtype", request_id)
+        payload = {key: value for key, value in raw.items() if key not in {"type", "id"}}
+        self._validate_payload_paths(payload, request_id)
+        return SDKControlRequest(request_id=request_id, subtype=subtype, payload=payload)
+
+    def _parse_legacy_initialize(self, raw: dict[str, Any], request_id: str | None) -> SDKControlRequest:
+        if not request_id:
+            raise ProcessFrameValidationError("invalid_frame", "initialize frame requires id.")
+        raw_options = raw.get("options")
+        options: dict[str, Any] = raw_options if isinstance(raw_options, dict) else {}
+        payload = dict(options)
+        self._validate_payload_paths(payload, request_id)
+        return SDKControlRequest(request_id=request_id, subtype="initialize", payload=payload)
+
+    def _parse_legacy_close(self, raw: dict[str, Any], request_id: str | None) -> SDKControlRequest:
+        if not request_id:
+            raise ProcessFrameValidationError("invalid_frame", "close frame requires id.")
+        return SDKControlRequest(request_id=request_id, subtype="close", payload={"subtype": "close"})
+
+    def _parse_user(self, raw: dict[str, Any], request_id: str | None) -> SDKUserMessage:
+        message = raw.get("message")
+        if not isinstance(message, dict):
+            raise ProcessFrameValidationError("invalid_frame", "user.message must be an object.", request_id)
+        if message.get("role") not in (None, "user"):
+            raise ProcessFrameValidationError("invalid_frame", "user.message.role must be user.", request_id)
+        metadata = _metadata(raw.get("metadata"), request_id)
+        cwd = _metadata_cwd(metadata, request_id)
+        return SDKUserMessage(
+            request_id=request_id,
+            session_id=_optional_string(raw.get("session_id")),
+            text=_content_text(message.get("content"), request_id),
+            metadata=metadata,
+            cwd=cwd,
+            parent_tool_use_id=_nullable_string(raw.get("parent_tool_use_id"), "parent_tool_use_id", request_id),
+            uuid=_optional_string(raw.get("uuid")),
+            timestamp=_optional_string(raw.get("timestamp")),
+        )
+
+    def _parse_legacy_user(self, raw: dict[str, Any], request_id: str | None) -> SDKUserMessage:
+        metadata = _metadata(raw.get("metadata"), request_id)
+        cwd = _metadata_cwd(metadata, request_id)
+        return SDKUserMessage(
+            request_id=request_id,
+            session_id=_optional_string(raw.get("session_id")),
+            text=_content_text(raw.get("content"), request_id),
+            metadata=metadata,
+            cwd=cwd,
+            parent_tool_use_id=_nullable_string(raw.get("parent_tool_use_id"), "parent_tool_use_id", request_id),
+            uuid=_optional_string(raw.get("uuid")),
+            timestamp=_optional_string(raw.get("timestamp")),
+        )
+
+    def _validate_payload_paths(self, payload: dict[str, Any], request_id: str | None) -> None:
+        cwd = payload.get("cwd")
+        if cwd is not None:
+            if not isinstance(cwd, str):
+                raise ProcessFrameValidationError("invalid_frame", "cwd must be a string.", request_id)
+            _validate_absolute_cwd(cwd, request_id)
+
+
+def _optional_string(value: Any) -> str | None:
+    return value if isinstance(value, str) and value else None
+
+
+def _nullable_string(value: Any, field_name: str, request_id: str | None) -> str | None:
+    if value is None:
+        return None
+    if isinstance(value, str):
+        return value
+    raise ProcessFrameValidationError("invalid_frame", f"{field_name} must be a string or null.", request_id)
+
+
+def _required_string(value: Any, field_name: str, request_id: str | None) -> str:
+    if not isinstance(value, str) or not value:
+        raise ProcessFrameValidationError("invalid_frame", f"{field_name} must be a non-empty string.", request_id)
+    return value
+
+
+def _metadata(value: Any, request_id: str | None) -> dict[str, Any]:
+    if value is None:
+        return {}
+    if not isinstance(value, dict):
+        raise ProcessFrameValidationError("invalid_frame", "metadata must be an object.", request_id)
+    return value
+
+
+def _metadata_cwd(metadata: dict[str, Any], request_id: str | None) -> str | None:
+    iac_code = metadata.get("iac_code")
+    if not isinstance(iac_code, dict):
+        return None
+    cwd = iac_code.get("cwd")
+    if cwd is None:
+        return None
+    if not isinstance(cwd, str):
+        raise ProcessFrameValidationError("invalid_frame", "metadata.iac_code.cwd must be a string.", request_id)
+    _validate_absolute_cwd(cwd, request_id)
+    return cwd
+
+
+def _validate_absolute_cwd(cwd: str, request_id: str | None) -> None:
+    if not os.path.isabs(os.path.expanduser(cwd)):
+        raise ProcessFrameValidationError("invalid_frame", "cwd must be an absolute path.", request_id)
+
+
+def _content_text(value: Any, request_id: str | None) -> str:
+    if isinstance(value, str):
+        return value
+    if not isinstance(value, list):
+        raise ProcessFrameValidationError("invalid_frame", "content must be text or text blocks.", request_id)
+    parts: list[str] = []
+    for block in value:
+        if not isinstance(block, dict) or block.get("type") != "text" or not isinstance(block.get("text"), str):
+            raise ProcessFrameValidationError("invalid_frame", "content blocks must be text blocks.", request_id)
+        parts.append(block["text"])
+    return "".join(parts)
diff --git a/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po
index 991fcd6c..8da1e1bf 100644
--- a/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po
@@ -707,6 +707,10 @@ msgstr "Nicht-interaktiver Modus: Einen Prompt ausführen und beenden"
 msgid "Output format: text, json, stream-json"
 msgstr "Ausgabeformat: text, json, stream-json"
 
+#: src/iac_code/cli/main.py
+msgid "Input format: text, stream-json"
+msgstr "Eingabeformat: text, stream-json"
+
 #: src/iac_code/cli/main.py
 msgid "Maximum agent turns in headless mode"
 msgstr "Maximale Agent-Runden im Headless-Modus"
@@ -772,6 +776,19 @@ msgstr ""
 "Error: --resume and --continue cannot be used together.Fehler: --resume "
 "und --continue können nicht gemeinsam verwendet werden."
 
+#: src/iac_code/cli/main.py
+#, python-brace-format
+msgid "Invalid --input-format '{}'. Valid values: text, stream-json"
+msgstr "Ungültiges --input-format '{}'. Gültige Werte: text, stream-json"
+
+#: src/iac_code/cli/main.py
+msgid "--prompt cannot be used with --input-format stream-json."
+msgstr "--prompt kann nicht mit --input-format stream-json verwendet werden."
+
+#: src/iac_code/cli/main.py
+msgid "--input-format stream-json requires --output-format stream-json."
+msgstr "--input-format stream-json erfordert --output-format stream-json."
+
 #: src/iac_code/cli/main.py
 #, python-brace-format
 msgid "Invalid --output-format '{}'. Valid values: {}"
@@ -6874,3 +6891,6 @@ msgstr "Öffnen einer nicht regulären Datei verweigert: {path}"
 #~ msgid "feature"
 #~ msgstr "Funktion"
 
+#~ msgid "Pipeline mode does not support --input-format stream-json."
+#~ msgstr "Pipeline-Modus unterstützt --input-format stream-json nicht."
+
diff --git a/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po
index d9b6b60c..2c1a7c11 100644
--- a/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po
@@ -711,6 +711,10 @@ msgstr "Modo no interactivo: ejecuta un único prompt y termina"
 msgid "Output format: text, json, stream-json"
 msgstr "Formato de salida: text, json, stream-json"
 
+#: src/iac_code/cli/main.py
+msgid "Input format: text, stream-json"
+msgstr "Formato de entrada: text, stream-json"
+
 #: src/iac_code/cli/main.py
 msgid "Maximum agent turns in headless mode"
 msgstr "Turnos máximos del agente en modo sin interfaz"
@@ -771,6 +775,19 @@ msgstr "Modo de permisos: default, accept_edits, bypass_permissions, dont_ask"
 msgid "Error: --resume and --continue cannot be used together."
 msgstr "Error: --resume y --continue no pueden usarse a la vez."
 
+#: src/iac_code/cli/main.py
+#, python-brace-format
+msgid "Invalid --input-format '{}'. Valid values: text, stream-json"
+msgstr "--input-format '{}' no válido. Valores válidos: text, stream-json"
+
+#: src/iac_code/cli/main.py
+msgid "--prompt cannot be used with --input-format stream-json."
+msgstr "--prompt no se puede usar con --input-format stream-json."
+
+#: src/iac_code/cli/main.py
+msgid "--input-format stream-json requires --output-format stream-json."
+msgstr "--input-format stream-json requiere --output-format stream-json."
+
 #: src/iac_code/cli/main.py
 #, python-brace-format
 msgid "Invalid --output-format '{}'. Valid values: {}"
@@ -6851,3 +6868,6 @@ msgstr "Se rechazó abrir un archivo no regular: {path}"
 #~ msgid "feature"
 #~ msgstr "función"
 
+#~ msgid "Pipeline mode does not support --input-format stream-json."
+#~ msgstr "El modo pipeline no admite --input-format stream-json."
+
diff --git a/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po
index 34f1c3d5..023b09b0 100644
--- a/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po
@@ -708,6 +708,10 @@ msgstr "Mode non interactif : exécuter un seul prompt puis quitter"
 msgid "Output format: text, json, stream-json"
 msgstr "Format de sortie : text, json, stream-json"
 
+#: src/iac_code/cli/main.py
+msgid "Input format: text, stream-json"
+msgstr "Format d'entrée : text, stream-json"
+
 #: src/iac_code/cli/main.py
 msgid "Maximum agent turns in headless mode"
 msgstr "Nombre maximal de tours d’agent en mode headless"
@@ -770,6 +774,19 @@ msgstr ""
 msgid "Error: --resume and --continue cannot be used together."
 msgstr "Erreur : --resume et --continue ne peuvent pas être utilisés ensemble."
 
+#: src/iac_code/cli/main.py
+#, python-brace-format
+msgid "Invalid --input-format '{}'. Valid values: text, stream-json"
+msgstr "--input-format '{}' invalide. Valeurs valides : text, stream-json"
+
+#: src/iac_code/cli/main.py
+msgid "--prompt cannot be used with --input-format stream-json."
+msgstr "--prompt ne peut pas être utilisé avec --input-format stream-json."
+
+#: src/iac_code/cli/main.py
+msgid "--input-format stream-json requires --output-format stream-json."
+msgstr "--input-format stream-json nécessite --output-format stream-json."
+
 #: src/iac_code/cli/main.py
 #, python-brace-format
 msgid "Invalid --output-format '{}'. Valid values: {}"
@@ -6872,3 +6889,6 @@ msgstr "Refus d’ouvrir un fichier non ordinaire : {path}"
 #~ msgid "feature"
 #~ msgstr "fonctionnalité"
 
+#~ msgid "Pipeline mode does not support --input-format stream-json."
+#~ msgstr "Le mode pipeline ne prend pas en charge --input-format stream-json."
+
diff --git a/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po
index 15156437..3176652d 100644
--- a/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po
@@ -670,6 +670,10 @@ msgstr "非対話モード：単一のプロンプトを実行して終了しま
 msgid "Output format: text, json, stream-json"
 msgstr "出力形式：text、json、stream-json"
 
+#: src/iac_code/cli/main.py
+msgid "Input format: text, stream-json"
+msgstr "入力形式: text、stream-json"
+
 #: src/iac_code/cli/main.py
 msgid "Maximum agent turns in headless mode"
 msgstr "ヘッドレスモードでの最大エージェンターン数"
@@ -726,6 +730,19 @@ msgstr "権限モード: default, accept_edits, bypass_permissions, dont_ask"
 msgid "Error: --resume and --continue cannot be used together."
 msgstr "エラー：--resume と --continue は同時に使用できません。"
 
+#: src/iac_code/cli/main.py
+#, python-brace-format
+msgid "Invalid --input-format '{}'. Valid values: text, stream-json"
+msgstr "無効な --input-format '{}' です。有効な値: text、stream-json"
+
+#: src/iac_code/cli/main.py
+msgid "--prompt cannot be used with --input-format stream-json."
+msgstr "--prompt は --input-format stream-json と同時に使用できません。"
+
+#: src/iac_code/cli/main.py
+msgid "--input-format stream-json requires --output-format stream-json."
+msgstr "--input-format stream-json には --output-format stream-json が必要です。"
+
 #: src/iac_code/cli/main.py
 #, python-brace-format
 msgid "Invalid --output-format '{}'. Valid values: {}"
@@ -6493,3 +6510,6 @@ msgstr "通常ファイルではないファイルを開くことを拒否しま
 #~ msgid "feature"
 #~ msgstr "機能"
 
+#~ msgid "Pipeline mode does not support --input-format stream-json."
+#~ msgstr "Pipeline モードは --input-format stream-json をサポートしていません。"
+
diff --git a/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po
index 8b2ae21a..45e33980 100644
--- a/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po
@@ -704,6 +704,10 @@ msgstr "Modo não interativo: executa um único prompt e encerra"
 msgid "Output format: text, json, stream-json"
 msgstr "Formato de saída: text, json, stream-json"
 
+#: src/iac_code/cli/main.py
+msgid "Input format: text, stream-json"
+msgstr "Formato de entrada: text, stream-json"
+
 #: src/iac_code/cli/main.py
 msgid "Maximum agent turns in headless mode"
 msgstr "Número máximo de passos do agent em modo headless"
@@ -764,6 +768,19 @@ msgstr "Modo de permissão: default, accept_edits, bypass_permissions, dont_ask"
 msgid "Error: --resume and --continue cannot be used together."
 msgstr "Erro: --resume e --continue não podem ser usados juntos."
 
+#: src/iac_code/cli/main.py
+#, python-brace-format
+msgid "Invalid --input-format '{}'. Valid values: text, stream-json"
+msgstr "--input-format inválido '{}'. Valores válidos: text, stream-json"
+
+#: src/iac_code/cli/main.py
+msgid "--prompt cannot be used with --input-format stream-json."
+msgstr "--prompt não pode ser usado com --input-format stream-json."
+
+#: src/iac_code/cli/main.py
+msgid "--input-format stream-json requires --output-format stream-json."
+msgstr "--input-format stream-json requer --output-format stream-json."
+
 #: src/iac_code/cli/main.py
 #, python-brace-format
 msgid "Invalid --output-format '{}'. Valid values: {}"
@@ -6790,3 +6807,6 @@ msgstr "Recusado abrir arquivo não regular: {path}"
 #~ msgid "feature"
 #~ msgstr "recurso"
 
+#~ msgid "Pipeline mode does not support --input-format stream-json."
+#~ msgstr "O modo pipeline não oferece suporte a --input-format stream-json."
+
diff --git a/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po
index bfa8aa09..20a214fd 100644
--- a/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po
@@ -664,6 +664,10 @@ msgstr "非交互模式：运行单个提示并退出"
 msgid "Output format: text, json, stream-json"
 msgstr "输出格式：text、json、stream-json"
 
+#: src/iac_code/cli/main.py
+msgid "Input format: text, stream-json"
+msgstr "输入格式：text、stream-json"
+
 #: src/iac_code/cli/main.py
 msgid "Maximum agent turns in headless mode"
 msgstr "无头模式下最大智能体轮次"
@@ -720,6 +724,19 @@ msgstr "权限模式：default, accept_edits, bypass_permissions, dont_ask"
 msgid "Error: --resume and --continue cannot be used together."
 msgstr "错误：--resume 和 --continue 不能同时使用。"
 
+#: src/iac_code/cli/main.py
+#, python-brace-format
+msgid "Invalid --input-format '{}'. Valid values: text, stream-json"
+msgstr "无效的 --input-format '{}'。有效值：text、stream-json"
+
+#: src/iac_code/cli/main.py
+msgid "--prompt cannot be used with --input-format stream-json."
+msgstr "--prompt 不能与 --input-format stream-json 同时使用。"
+
+#: src/iac_code/cli/main.py
+msgid "--input-format stream-json requires --output-format stream-json."
+msgstr "--input-format stream-json 需要 --output-format stream-json。"
+
 #: src/iac_code/cli/main.py
 #, python-brace-format
 msgid "Invalid --output-format '{}'. Valid values: {}"
@@ -6396,3 +6413,6 @@ msgstr "拒绝打开非常规文件：{path}"
 #~ msgid "feature"
 #~ msgstr "功能"
 
+#~ msgid "Pipeline mode does not support --input-format stream-json."
+#~ msgstr "Pipeline 模式不支持 --input-format stream-json。"
+
diff --git a/src/iac_code/services/permissions/trusted_roots.py b/src/iac_code/services/permissions/trusted_roots.py
index 91175aca..e580ee58 100644
--- a/src/iac_code/services/permissions/trusted_roots.py
+++ b/src/iac_code/services/permissions/trusted_roots.py
@@ -29,8 +29,14 @@ def build_session_trusted_read_directories(
         str(config_dir / "tool-results" / session_id),
         str(config_dir / "image-cache" / session_id),
     ]
-    if session_dir is not None:
-        session_paths = SessionPaths.from_session_dir(session_dir)
+    if isinstance(session_dir, Path):
+        session_path = session_dir
+    elif isinstance(session_dir, str):
+        session_path = Path(session_dir)
+    else:
+        session_path = None
+    if session_path is not None:
+        session_paths = SessionPaths.from_session_dir(session_path)
         for path in (session_paths.tool_results_dir, session_paths.image_cache_dir):
             try:
                 roots.append(str(ensure_session_owned_dir(session_paths.session_dir, path)))
diff --git a/src/iac_code/ui/repl.py b/src/iac_code/ui/repl.py
index 308a9f61..42469fbe 100644
--- a/src/iac_code/ui/repl.py
+++ b/src/iac_code/ui/repl.py
@@ -727,7 +727,7 @@ def _raw_session_dir_for_trusted_roots(self, session_id: str | None = None):
             return None
         if isinstance(raw_session_dir, Path):
             return raw_session_dir
-        if isinstance(raw_session_dir, (str, os.PathLike)):
+        if isinstance(raw_session_dir, str):
             return Path(raw_session_dir)
         return None
 
diff --git a/tests/cli/test_process_cli.py b/tests/cli/test_process_cli.py
new file mode 100644
index 00000000..92d7153b
--- /dev/null
+++ b/tests/cli/test_process_cli.py
@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+from typer.testing import CliRunner
+
+from iac_code.cli.main import app
+
+
+def test_help_includes_input_format_option() -> None:
+    result = CliRunner().invoke(app, ["--help"])
+
+    assert result.exit_code == 0
+    assert "--input-format" in result.output
+
+
+def test_process_mode_rejects_prompt_combination() -> None:
+    result = CliRunner().invoke(
+        app,
+        ["--input-format", "stream-json", "--output-format", "stream-json", "--prompt", "hello"],
+    )
+
+    assert result.exit_code == 1
+    assert "--prompt cannot be used with --input-format stream-json" in result.output
+
+
+def test_process_mode_requires_stream_json_output() -> None:
+    result = CliRunner().invoke(app, ["--input-format", "stream-json", "--output-format", "json"])
+
+    assert result.exit_code == 1
+    assert "--input-format stream-json requires --output-format stream-json" in result.output
+
+
+def test_process_mode_rejects_invalid_input_format() -> None:
+    result = CliRunner().invoke(app, ["--input-format", "yaml"])
+
+    assert result.exit_code == 1
+    assert "Invalid --input-format 'yaml'" in result.output
+
+
+def test_process_mode_invokes_runner_in_pipeline_mode(monkeypatch, tmp_path) -> None:
+    captured = {}
+
+    class FakeProcessModeRunner:
+        def __init__(self, options) -> None:
+            captured["options"] = options
+
+        async def run(self) -> int:
+            captured["ran"] = True
+            return 0
+
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("IAC_CODE_MODE", "pipeline")
+    monkeypatch.setattr("iac_code.cli.process_mode.ProcessModeRunner", FakeProcessModeRunner)
+
+    result = CliRunner().invoke(app, ["--input-format", "stream-json", "--output-format", "stream-json"])
+
+    assert result.exit_code == 0
+    assert captured["ran"] is True
+    assert captured["options"].cwd == str(tmp_path)
+    assert captured["options"].run_mode == "pipeline"
+
+
+def test_process_mode_invokes_runner(monkeypatch, tmp_path) -> None:
+    captured = {}
+
+    class FakeProcessModeRunner:
+        def __init__(self, options) -> None:
+            captured["options"] = options
+
+        async def run(self) -> int:
+            captured["ran"] = True
+            return 0
+
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("IAC_CODE_MODE", "normal")
+    monkeypatch.setattr("iac_code.cli.process_mode.ProcessModeRunner", FakeProcessModeRunner)
+
+    result = CliRunner().invoke(app, ["--input-format", "stream-json", "--output-format", "stream-json"])
+
+    assert result.exit_code == 0
+    assert captured["ran"] is True
+    assert captured["options"].cwd == str(tmp_path)
+
+
+def test_process_mode_unknown_iac_code_mode_falls_back_to_normal(monkeypatch, tmp_path) -> None:
+    captured = {}
+
+    class FakeProcessModeRunner:
+        def __init__(self, options) -> None:
+            captured["options"] = options
+
+        async def run(self) -> int:
+            captured["ran"] = True
+            return 0
+
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("IAC_CODE_MODE", "unexpected")
+    monkeypatch.setattr("iac_code.cli.process_mode.ProcessModeRunner", FakeProcessModeRunner)
+
+    result = CliRunner().invoke(app, ["--input-format", "stream-json", "--output-format", "stream-json"])
+
+    assert result.exit_code == 0
+    assert captured["ran"] is True
+    assert captured["options"].cwd == str(tmp_path)
diff --git a/tests/cli/test_process_events.py b/tests/cli/test_process_events.py
new file mode 100644
index 00000000..c3ed6e8b
--- /dev/null
+++ b/tests/cli/test_process_events.py
@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import json
+
+from iac_code.cli.process_events import ProcessErrorMapper, ProcessEventSerializer
+from iac_code.providers.manager import ProviderNotConfiguredError
+from iac_code.types.stream_events import ErrorEvent, TextDeltaEvent, ToolInputDeltaEvent, ToolResultEvent
+
+
+def test_process_event_serializer_reuses_stream_json_shape_for_text_delta() -> None:
+    serializer = ProcessEventSerializer()
+
+    assert serializer.serialize(TextDeltaEvent(text="hello")) == {"type": "text_delta", "text": "hello"}
+
+
+def test_process_event_serializer_hides_partial_tool_input() -> None:
+    serializer = ProcessEventSerializer()
+
+    assert serializer.serialize(ToolInputDeltaEvent(tool_use_id="tool-1", partial_json='{"secret":"value"}')) == {
+        "type": "tool_input_delta",
+        "tool_use_id": "tool-1",
+        "partial_json_length": 18,
+    }
+
+
+def test_process_event_serializer_sanitizes_tool_result() -> None:
+    serializer = ProcessEventSerializer()
+
+    event = ToolResultEvent(
+        tool_use_id="tool-1",
+        tool_name="bash",
+        result="failed with token sk-live12345 at /Users/alice/.iac-code/settings.yml",
+        is_error=True,
+    )
+
+    rendered = json.dumps(serializer.serialize(event), ensure_ascii=False)
+    assert "sk-live12345" not in rendered
+    assert "/Users/alice" not in rendered
+    assert "settings.yml" not in rendered
+
+
+def test_process_error_mapper_returns_stable_provider_code() -> None:
+    mapper = ProcessErrorMapper()
+    payload = mapper.from_exception(ProviderNotConfiguredError("DashScope provider not configured"))
+
+    assert payload.code == "provider_not_configured"
+    assert payload.retryable is False
+    assert "DashScope" in payload.message
+
+
+def test_process_error_mapper_preserves_stream_error_event_retryability() -> None:
+    mapper = ProcessErrorMapper()
+    payload = mapper.from_event(ErrorEvent(error="temporary", is_retryable=True, error_id="err-1"))
+
+    assert payload.code == "stream_error"
+    assert payload.retryable is True
+    assert payload.error_id == "err-1"
diff --git a/tests/cli/test_process_mode.py b/tests/cli/test_process_mode.py
new file mode 100644
index 00000000..520f3b7a
--- /dev/null
+++ b/tests/cli/test_process_mode.py
@@ -0,0 +1,597 @@
+from __future__ import annotations
+
+import asyncio
+import io
+import json
+import os
+from collections.abc import AsyncIterator
+
+import pytest
+
+from iac_code.cli.process_mode import (
+    PipelineProcessContextSnapshot,
+    PipelineProcessContextStore,
+    PipelineProcessRuntimeController,
+    ProcessModeOptions,
+    ProcessModeRunner,
+    ProcessRuntimeController,
+    ProcessSessionLock,
+)
+from iac_code.cli.process_protocol import SDKControlRequest, SDKUserMessage
+from iac_code.pipeline.engine.events import PipelineEvent, PipelineEventType
+from iac_code.types.stream_events import MessageEndEvent, TextDeltaEvent, Usage
+
+
+class FakeRuntimeController:
+    def __init__(self, *, block_turn: bool = False) -> None:
+        self.model = "model-a"
+        self.session_id = "generated-session"
+        self.turns: list[tuple[str, str]] = []
+        self.closed = False
+        self.block_turn = block_turn
+        self.turn_started = asyncio.Event()
+        self.release_turn = asyncio.Event()
+
+    async def initialize(self, frame) -> dict:
+        if frame.payload.get("model"):
+            self.model = frame.payload["model"]
+        return {"protocol_version": "1.0", "capabilities": ["user", "interrupt", "set_model", "end_session", "close"]}
+
+    def set_model(self, model: str) -> None:
+        self.model = model
+
+    async def run_turn(self, frame) -> AsyncIterator[object]:
+        self.turns.append((self.model, frame.text))
+        self.turn_started.set()
+        if self.block_turn:
+            await self.release_turn.wait()
+        yield TextDeltaEvent(text=f"echo:{frame.text}")
+        yield MessageEndEvent(stop_reason="end_turn", usage=Usage(input_tokens=1, output_tokens=1))
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+class FakePipelineRuntime:
+    provider_manager = object()
+    tool_registry = object()
+    command_registry = object()
+
+    async def aclose(self) -> None:
+        return None
+
+
+class FakePipeline:
+    def __init__(self) -> None:
+        self.sidecar_status: str | None = None
+        self.inputs: list[tuple[str, str]] = []
+
+    async def run(self, user_input):
+        self.inputs.append(("run", user_input.display_text))
+        yield PipelineEvent(
+            type=PipelineEventType.PIPELINE_STARTED,
+            step_id=None,
+            timestamp=1.0,
+            data={"input": user_input.display_text},
+        )
+        self.sidecar_status = "waiting_input"
+        yield PipelineEvent(
+            type=PipelineEventType.USER_INPUT_REQUIRED,
+            step_id=None,
+            timestamp=2.0,
+            data={"prompt": "continue?"},
+        )
+
+    async def resume(self, user_input):
+        self.inputs.append(("resume", user_input.display_text))
+        self.sidecar_status = "running"
+        yield PipelineEvent(
+            type=PipelineEventType.USER_INPUT_RECEIVED,
+            step_id=None,
+            timestamp=3.0,
+            data={"input": user_input.display_text},
+        )
+        self.sidecar_status = "completed"
+        yield PipelineEvent(
+            type=PipelineEventType.PIPELINE_COMPLETED,
+            step_id=None,
+            timestamp=4.0,
+            data={},
+        )
+
+
+def _load_output(stream: io.StringIO) -> list[dict]:
+    return [json.loads(line) for line in stream.getvalue().splitlines() if line.strip()]
+
+
+@pytest.mark.asyncio
+async def test_pipeline_process_runner_starts_waits_and_resumes(tmp_path) -> None:
+    pipeline = FakePipeline()
+    controller = PipelineProcessRuntimeController(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path), run_mode="pipeline"),
+        agent_runtime_factory=lambda request: FakePipelineRuntime(),
+        pipeline_factory=lambda request: pipeline,
+    )
+    metadata = {
+        "iac_code": {
+            "contextId": "ctx-1",
+            "taskId": "task-1",
+            "iacCodeSessionId": "iac-session-1",
+        }
+    }
+    stdin = io.StringIO(
+        "\n".join(
+            [
+                json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}}),
+                json.dumps(
+                    {
+                        "type": "user",
+                        "request_id": "req-start",
+                        "session_id": "sdk-session-1",
+                        "metadata": metadata,
+                        "message": {"role": "user", "content": "start"},
+                    }
+                ),
+                json.dumps(
+                    {
+                        "type": "user",
+                        "request_id": "req-resume",
+                        "session_id": "sdk-session-1",
+                        "metadata": metadata,
+                        "message": {"role": "user", "content": "continue"},
+                    }
+                ),
+            ]
+        )
+        + "\n"
+    )
+    stdout = io.StringIO()
+
+    exit_code = await ProcessModeRunner(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path), run_mode="pipeline"),
+        input_stream=stdin,
+        output_stream=stdout,
+        runtime_controller=controller,
+    ).run()
+
+    assert exit_code == 0
+    assert pipeline.inputs == [("run", "start"), ("resume", "continue")]
+    lines = _load_output(stdout)
+    init = lines[0]
+    assert init["response"]["subtype"] == "success"
+    assert "pipeline_resume" in init["response"]["response"]["capabilities"]
+
+    stream_events = [line["event"] for line in lines if line["type"] == "stream_event"]
+    assert [event["eventType"] for event in stream_events] == [
+        "pipeline_started",
+        "input_required",
+        "input_received",
+        "pipeline_completed",
+    ]
+    assert all(event["type"] == "pipeline_event" for event in stream_events)
+    assert all(event["contextId"] == "ctx-1" for event in stream_events)
+    assert all(event["taskId"] == "task-1" for event in stream_events)
+
+    results = [line for line in lines if line["type"] == "result"]
+    assert results[0]["request_id"] == "req-start"
+    assert results[0]["stop_reason"] == "input_required"
+    assert results[0]["pipeline"]["status"] == "input_required"
+    assert results[0]["pipeline"]["sidecarStatus"] == "waiting_input"
+    assert results[1]["request_id"] == "req-resume"
+    assert results[1]["stop_reason"] == "end_turn"
+    assert results[1]["pipeline"]["status"] == "completed"
+    assert results[1]["pipeline"]["sidecarStatus"] == "completed"
+
+
+@pytest.mark.asyncio
+async def test_pipeline_process_runner_returns_recoverable_task_when_task_id_is_missing(tmp_path) -> None:
+    context_store = PipelineProcessContextStore(tmp_path / "contexts")
+    context_store.save(
+        PipelineProcessContextSnapshot(
+            context_id="ctx-1",
+            task_id="task-1",
+            iac_code_session_id="iac-session-1",
+            cwd=str(tmp_path),
+            sidecar_status="waiting_input",
+            active_task_id="task-1",
+        )
+    )
+    controller = PipelineProcessRuntimeController(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path), run_mode="pipeline"),
+        agent_runtime_factory=lambda request: FakePipelineRuntime(),
+        pipeline_factory=lambda request: FakePipeline(),
+        context_store=context_store,
+    )
+    stdin = io.StringIO(
+        "\n".join(
+            [
+                json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}}),
+                json.dumps(
+                    {
+                        "type": "user",
+                        "request_id": "req-resume",
+                        "metadata": {"iac_code": {"contextId": "ctx-1"}},
+                        "message": {"role": "user", "content": "continue"},
+                    }
+                ),
+            ]
+        )
+        + "\n"
+    )
+    stdout = io.StringIO()
+
+    exit_code = await ProcessModeRunner(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path), run_mode="pipeline"),
+        input_stream=stdin,
+        output_stream=stdout,
+        runtime_controller=controller,
+    ).run()
+
+    assert exit_code == 0
+    lines = _load_output(stdout)
+    error = next(line for line in lines if line["type"] == "error")
+    assert error["request_id"] == "req-resume"
+    assert error["error"]["code"] == "pipeline_task_required"
+    assert error["error"]["retryable"] is True
+    assert error["error"]["data"] == {
+        "contextId": "ctx-1",
+        "recoverableTaskId": "task-1",
+        "sidecarStatus": "waiting_input",
+    }
+
+
+@pytest.mark.asyncio
+async def test_process_runner_handles_initialize_user_and_eof(tmp_path) -> None:
+    controller = FakeRuntimeController()
+    stdin = io.StringIO(
+        "\n".join(
+            [
+                json.dumps(
+                    {
+                        "type": "control_request",
+                        "request_id": "req-init",
+                        "request": {"subtype": "initialize", "cwd": str(tmp_path), "model": "model-a"},
+                    }
+                ),
+                json.dumps({"type": "user", "session_id": "session-1", "message": {"role": "user", "content": "hi"}}),
+            ]
+        )
+        + "\n"
+    )
+    stdout = io.StringIO()
+
+    exit_code = await ProcessModeRunner(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path)),
+        input_stream=stdin,
+        output_stream=stdout,
+        runtime_controller=controller,
+    ).run()
+
+    assert exit_code == 0
+    assert controller.turns == [("model-a", "hi")]
+    lines = _load_output(stdout)
+    assert lines[0]["type"] == "control_response"
+    assert lines[0]["response"]["subtype"] == "success"
+    assert lines[0]["response"]["request_id"] == "req-init"
+    assert lines[1]["type"] == "stream_event"
+    assert lines[1]["session_id"] == "session-1"
+    assert lines[1]["event"] == {"type": "text_delta", "text": "echo:hi"}
+    assert lines[2]["type"] == "stream_event"
+    assert lines[2]["event"]["type"] == "message_end"
+    assert lines[3]["type"] == "result"
+    assert lines[3]["subtype"] == "success"
+    assert lines[3]["is_error"] is False
+    assert lines[3]["result"] == "echo:hi"
+    assert lines[3]["stop_reason"] == "end_turn"
+    assert lines[3]["usage"]["input_tokens"] == 1
+    assert lines[3]["duration_api_ms"] == 0
+    assert lines[3]["num_turns"] == 1
+    assert lines[3]["modelUsage"] == {}
+
+
+@pytest.mark.asyncio
+async def test_process_runner_rejects_user_before_initialize(tmp_path) -> None:
+    controller = FakeRuntimeController()
+    stdin = io.StringIO(json.dumps({"type": "user", "message": {"role": "user", "content": "hi"}}) + "\n")
+    stdout = io.StringIO()
+
+    exit_code = await ProcessModeRunner(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path)),
+        input_stream=stdin,
+        output_stream=stdout,
+        runtime_controller=controller,
+    ).run()
+
+    assert exit_code == 1
+    lines = _load_output(stdout)
+    assert lines == [
+        {
+            "type": "error",
+            "request_id": None,
+            "error": {
+                "code": "not_initialized",
+                "message": "initialize is required before user messages",
+                "retryable": False,
+            },
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_process_runner_reports_turn_active_and_close_cancels_turn(tmp_path) -> None:
+    controller = FakeRuntimeController(block_turn=True)
+    stdin = io.StringIO(
+        "\n".join(
+            [
+                json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}}),
+                json.dumps(
+                    {"type": "user", "session_id": "session-1", "message": {"role": "user", "content": "first"}}
+                ),
+                json.dumps(
+                    {"type": "user", "session_id": "session-1", "message": {"role": "user", "content": "second"}}
+                ),
+                json.dumps(
+                    {"type": "control_request", "request_id": "req-close", "request": {"subtype": "end_session"}}
+                ),
+            ]
+        )
+        + "\n"
+    )
+    stdout = io.StringIO()
+
+    exit_code = await ProcessModeRunner(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path)),
+        input_stream=stdin,
+        output_stream=stdout,
+        runtime_controller=controller,
+    ).run()
+
+    assert exit_code == 0
+    lines = _load_output(stdout)
+    assert any(line["type"] == "error" and line["error"]["code"] == "turn_active" for line in lines)
+    canceled = [line for line in lines if line["type"] == "result" and line["subtype"] == "error_during_execution"]
+    assert len(canceled) == 1
+    assert canceled[0]["is_error"] is True
+    assert canceled[0]["stop_reason"] == "cancelled"
+    assert any(
+        line["type"] == "control_response"
+        and line["response"]["request_id"] == "req-close"
+        and line["response"]["subtype"] == "success"
+        for line in lines
+    )
+
+
+@pytest.mark.asyncio
+async def test_process_runner_interrupt_cancels_active_turn(tmp_path) -> None:
+    controller = FakeRuntimeController(block_turn=True)
+    stdin = io.StringIO(
+        "\n".join(
+            [
+                json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}}),
+                json.dumps(
+                    {"type": "user", "session_id": "session-1", "message": {"role": "user", "content": "first"}}
+                ),
+                json.dumps(
+                    {
+                        "type": "control_request",
+                        "request_id": "req-interrupt",
+                        "request": {"subtype": "interrupt"},
+                    }
+                ),
+            ]
+        )
+        + "\n"
+    )
+    stdout = io.StringIO()
+
+    exit_code = await ProcessModeRunner(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path)),
+        input_stream=stdin,
+        output_stream=stdout,
+        runtime_controller=controller,
+    ).run()
+
+    assert exit_code == 0
+    lines = _load_output(stdout)
+    assert any(
+        line["type"] == "control_response"
+        and line["response"]["request_id"] == "req-interrupt"
+        and line["response"]["subtype"] == "success"
+        for line in lines
+    )
+    assert any(line["type"] == "result" and line["subtype"] == "error_during_execution" for line in lines)
+
+
+@pytest.mark.asyncio
+async def test_process_runner_set_model_affects_next_turn(tmp_path) -> None:
+    controller = FakeRuntimeController()
+    stdin = io.StringIO(
+        "\n".join(
+            [
+                json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}}),
+                json.dumps(
+                    {
+                        "type": "control_request",
+                        "request_id": "req-model",
+                        "request": {"subtype": "set_model", "model": "model-b"},
+                    }
+                ),
+                json.dumps(
+                    {"type": "user", "session_id": "session-1", "message": {"role": "user", "content": "after"}}
+                ),
+            ]
+        )
+        + "\n"
+    )
+    stdout = io.StringIO()
+
+    exit_code = await ProcessModeRunner(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path)),
+        input_stream=stdin,
+        output_stream=stdout,
+        runtime_controller=controller,
+    ).run()
+
+    assert exit_code == 0
+    assert controller.turns == [("model-b", "after")]
+    lines = _load_output(stdout)
+    assert any(
+        line["type"] == "control_response"
+        and line["response"]["request_id"] == "req-model"
+        and line["response"]["subtype"] == "success"
+        for line in lines
+    )
+
+
+@pytest.mark.asyncio
+async def test_runtime_controller_uses_initialize_cwd_as_default(monkeypatch, tmp_path) -> None:
+    init_cwd = tmp_path / "init"
+    init_cwd.mkdir()
+    captured_cwds: list[str] = []
+
+    class FakeAgentLoop:
+        async def run_streaming(self, prompt: str) -> AsyncIterator[object]:
+            yield TextDeltaEvent(text=prompt)
+            yield MessageEndEvent(stop_reason="end_turn", usage=Usage())
+
+    class FakeRuntime:
+        session_id = "generated-session"
+        agent_loop = FakeAgentLoop()
+
+    def fake_create_agent_runtime(options) -> FakeRuntime:
+        captured_cwds.append(options.cwd)
+        return FakeRuntime()
+
+    monkeypatch.setattr("iac_code.services.agent_factory.create_agent_runtime", fake_create_agent_runtime)
+    controller = ProcessRuntimeController(ProcessModeOptions(model="model-a", cwd=str(tmp_path)))
+
+    response = await controller.initialize(
+        SDKControlRequest(
+            request_id="req-init",
+            subtype="initialize",
+            payload={"subtype": "initialize", "cwd": str(init_cwd)},
+        )
+    )
+    events = [
+        event async for event in controller.run_turn(SDKUserMessage(request_id=None, session_id=None, text="hello"))
+    ]
+
+    assert response["cwd"] == str(init_cwd)
+    assert captured_cwds == [str(init_cwd)]
+    assert isinstance(events[0], TextDeltaEvent)
+
+
+@pytest.mark.asyncio
+async def test_runtime_controller_user_cwd_overrides_initialize_cwd(monkeypatch, tmp_path) -> None:
+    init_cwd = tmp_path / "init"
+    message_cwd = tmp_path / "message"
+    init_cwd.mkdir()
+    message_cwd.mkdir()
+    captured_cwds: list[str] = []
+
+    class FakeAgentLoop:
+        async def run_streaming(self, prompt: str) -> AsyncIterator[object]:
+            yield TextDeltaEvent(text=prompt)
+            yield MessageEndEvent(stop_reason="end_turn", usage=Usage())
+
+    class FakeRuntime:
+        session_id = "generated-session"
+        agent_loop = FakeAgentLoop()
+
+    def fake_create_agent_runtime(options) -> FakeRuntime:
+        captured_cwds.append(options.cwd)
+        return FakeRuntime()
+
+    monkeypatch.setattr("iac_code.services.agent_factory.create_agent_runtime", fake_create_agent_runtime)
+    controller = ProcessRuntimeController(ProcessModeOptions(model="model-a", cwd=str(tmp_path)))
+
+    await controller.initialize(
+        SDKControlRequest(
+            request_id="req-init",
+            subtype="initialize",
+            payload={"subtype": "initialize", "cwd": str(init_cwd)},
+        )
+    )
+    _ = [
+        event
+        async for event in controller.run_turn(
+            SDKUserMessage(request_id=None, session_id=None, text="hello", cwd=str(message_cwd))
+        )
+    ]
+
+    assert captured_cwds == [str(message_cwd)]
+
+
+def test_process_session_lock_rejects_concurrent_holder(tmp_path) -> None:
+    first = ProcessSessionLock(cwd=str(tmp_path), session_id="session-1")
+    second = ProcessSessionLock(cwd=str(tmp_path), session_id="session-1")
+
+    assert first.acquire(blocking=False) is True
+    try:
+        assert second.acquire(blocking=False) is False
+    finally:
+        first.release()
+
+    assert second.acquire(blocking=False) is True
+    second.release()
+
+
+@pytest.mark.asyncio
+async def test_process_runner_applies_environment_updates(monkeypatch, tmp_path) -> None:
+    monkeypatch.delenv("IAC_CODE_PROCESS_TEST_ENV", raising=False)
+    controller = FakeRuntimeController()
+    stdin = io.StringIO(
+        "\n".join(
+            [
+                json.dumps({"type": "update_environment_variables", "variables": {"IAC_CODE_PROCESS_TEST_ENV": "1"}}),
+                json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}}),
+            ]
+        )
+        + "\n"
+    )
+    stdout = io.StringIO()
+
+    exit_code = await ProcessModeRunner(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path)),
+        input_stream=stdin,
+        output_stream=stdout,
+        runtime_controller=controller,
+    ).run()
+
+    assert exit_code == 0
+    assert os.environ["IAC_CODE_PROCESS_TEST_ENV"] == "1"
+    lines = _load_output(stdout)
+    assert len(lines) == 1
+    assert lines[0]["type"] == "control_response"
+
+
+@pytest.mark.asyncio
+async def test_process_runner_ignores_control_response_frames(tmp_path) -> None:
+    controller = FakeRuntimeController()
+    stdin = io.StringIO(
+        "\n".join(
+            [
+                json.dumps(
+                    {
+                        "type": "control_response",
+                        "response": {"subtype": "success", "request_id": "permission-1", "response": {"ok": True}},
+                    }
+                ),
+                json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}}),
+            ]
+        )
+        + "\n"
+    )
+    stdout = io.StringIO()
+
+    exit_code = await ProcessModeRunner(
+        ProcessModeOptions(model="model-a", cwd=str(tmp_path)),
+        input_stream=stdin,
+        output_stream=stdout,
+        runtime_controller=controller,
+    ).run()
+
+    assert exit_code == 0
+    lines = _load_output(stdout)
+    assert len(lines) == 1
+    assert lines[0]["response"]["request_id"] == "req-init"
diff --git a/tests/cli/test_process_protocol.py b/tests/cli/test_process_protocol.py
new file mode 100644
index 00000000..4c27a10e
--- /dev/null
+++ b/tests/cli/test_process_protocol.py
@@ -0,0 +1,191 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from iac_code.cli.process_protocol import (
+    ProcessFrameParser,
+    ProcessFrameValidationError,
+    SDKControlRequest,
+    SDKControlResponse,
+    SDKUpdateEnvironmentVariables,
+    SDKUserMessage,
+)
+
+
+def test_parse_claude_style_initialize_control_request(tmp_path) -> None:
+    parser = ProcessFrameParser()
+    frame = parser.parse_line(
+        json.dumps(
+            {
+                "type": "control_request",
+                "request_id": "req-1",
+                "request": {
+                    "subtype": "initialize",
+                    "cwd": str(tmp_path),
+                    "model": "qwen3.7-max",
+                    "max_turns": 10,
+                },
+            }
+        )
+    )
+
+    assert isinstance(frame, SDKControlRequest)
+    assert frame.request_id == "req-1"
+    assert frame.subtype == "initialize"
+    assert frame.payload["cwd"] == str(tmp_path)
+    assert frame.payload["model"] == "qwen3.7-max"
+    assert frame.payload["max_turns"] == 10
+
+
+def test_parse_user_frame_with_text_content() -> None:
+    parser = ProcessFrameParser()
+    frame = parser.parse_line(
+        json.dumps(
+            {
+                "type": "user",
+                "session_id": "session-1",
+                "message": {"role": "user", "content": [{"type": "text", "text": "hello"}]},
+                "metadata": {"iac_code": {"iac_code_model": "qwen3.6-plus"}},
+            }
+        )
+    )
+
+    assert isinstance(frame, SDKUserMessage)
+    assert frame.session_id == "session-1"
+    assert frame.text == "hello"
+    assert frame.metadata["iac_code"]["iac_code_model"] == "qwen3.6-plus"
+
+
+def test_parse_legacy_user_message_alias(tmp_path) -> None:
+    parser = ProcessFrameParser()
+    frame = parser.parse_line(
+        json.dumps(
+            {
+                "type": "user_message",
+                "id": "legacy-1",
+                "session_id": "session-legacy",
+                "content": [{"type": "text", "text": "legacy hello"}],
+                "metadata": {"iac_code": {"cwd": str(tmp_path)}},
+            }
+        )
+    )
+
+    assert isinstance(frame, SDKUserMessage)
+    assert frame.request_id == "legacy-1"
+    assert frame.session_id == "session-legacy"
+    assert frame.text == "legacy hello"
+    assert frame.cwd == str(tmp_path)
+
+
+def test_parse_legacy_control_alias() -> None:
+    parser = ProcessFrameParser()
+    frame = parser.parse_line(json.dumps({"type": "control", "id": "legacy-2", "subtype": "interrupt"}))
+
+    assert isinstance(frame, SDKControlRequest)
+    assert frame.request_id == "legacy-2"
+    assert frame.subtype == "interrupt"
+
+
+def test_parse_claude_style_control_response() -> None:
+    parser = ProcessFrameParser()
+    frame = parser.parse_line(
+        json.dumps(
+            {
+                "type": "control_response",
+                "response": {
+                    "subtype": "success",
+                    "request_id": "req-permission",
+                    "response": {"behavior": "allow"},
+                },
+            }
+        )
+    )
+
+    assert isinstance(frame, SDKControlResponse)
+    assert frame.request_id == "req-permission"
+    assert frame.subtype == "success"
+    assert frame.payload["response"] == {"behavior": "allow"}
+
+
+def test_parse_keep_alive_as_noop() -> None:
+    parser = ProcessFrameParser()
+
+    assert parser.parse_line(json.dumps({"type": "keep_alive"})) is None
+
+
+def test_parse_update_environment_variables() -> None:
+    parser = ProcessFrameParser()
+    frame = parser.parse_line(
+        json.dumps({"type": "update_environment_variables", "variables": {"IAC_CODE_TEST_ENV": "1"}})
+    )
+
+    assert isinstance(frame, SDKUpdateEnvironmentVariables)
+    assert frame.variables == {"IAC_CODE_TEST_ENV": "1"}
+
+
+def test_rejects_malformed_json() -> None:
+    parser = ProcessFrameParser()
+
+    with pytest.raises(ProcessFrameValidationError) as exc_info:
+        parser.parse_line("{not-json")
+
+    assert exc_info.value.code == "invalid_json"
+    assert exc_info.value.request_id is None
+
+
+def test_rejects_unknown_type() -> None:
+    parser = ProcessFrameParser()
+
+    with pytest.raises(ProcessFrameValidationError) as exc_info:
+        parser.parse_line(json.dumps({"type": "unknown", "id": "req-bad"}))
+
+    assert exc_info.value.code == "invalid_frame"
+    assert exc_info.value.request_id == "req-bad"
+
+
+def test_rejects_relative_cwd_in_initialize() -> None:
+    parser = ProcessFrameParser()
+
+    with pytest.raises(ProcessFrameValidationError) as exc_info:
+        parser.parse_line(
+            json.dumps(
+                {
+                    "type": "control_request",
+                    "request_id": "req-relative",
+                    "request": {"subtype": "initialize", "cwd": "relative/path"},
+                }
+            )
+        )
+
+    assert exc_info.value.code == "invalid_frame"
+    assert exc_info.value.request_id == "req-relative"
+    assert "cwd" in exc_info.value.message
+
+
+def test_rejects_non_text_content_block() -> None:
+    parser = ProcessFrameParser()
+
+    with pytest.raises(ProcessFrameValidationError) as exc_info:
+        parser.parse_line(
+            json.dumps(
+                {
+                    "type": "user",
+                    "session_id": "session-1",
+                    "message": {"role": "user", "content": [{"type": "image", "data": "..."}]},
+                }
+            )
+        )
+
+    assert exc_info.value.code == "invalid_frame"
+    assert "text" in exc_info.value.message
+
+
+def test_rejects_update_environment_variables_with_non_string_value() -> None:
+    parser = ProcessFrameParser()
+
+    with pytest.raises(ProcessFrameValidationError) as exc_info:
+        parser.parse_line(json.dumps({"type": "update_environment_variables", "variables": {"A": 1}}))
+
+    assert exc_info.value.code == "invalid_frame"
diff --git a/tests/cli/test_process_subprocess.py b/tests/cli/test_process_subprocess.py
new file mode 100644
index 00000000..0acebef2
--- /dev/null
+++ b/tests/cli/test_process_subprocess.py
@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+def _run_process_cli(*, config_dir: Path, stdin: str, mode: str | None = None) -> subprocess.CompletedProcess[str]:
+    env = os.environ.copy()
+    env["IAC_CODE_CONFIG_DIR"] = str(config_dir)
+    if mode is None:
+        env.pop("IAC_CODE_MODE", None)
+    else:
+        env["IAC_CODE_MODE"] = mode
+    return subprocess.run(
+        ["uv", "run", "iac-code", "--input-format", "stream-json", "--output-format", "stream-json"],
+        cwd=REPO_ROOT,
+        env=env,
+        input=stdin,
+        text=True,
+        capture_output=True,
+        timeout=30,
+        check=False,
+    )
+
+
+def _stdout_frames(result: subprocess.CompletedProcess[str]) -> list[dict]:
+    return [json.loads(line) for line in result.stdout.splitlines() if line.strip()]
+
+
+def test_process_subprocess_normal_mode_initialize_and_end_session(tmp_path) -> None:
+    result = _run_process_cli(
+        config_dir=tmp_path,
+        mode="normal",
+        stdin="\n".join(
+            [
+                json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}}),
+                json.dumps({"type": "control_request", "request_id": "req-end", "request": {"subtype": "end_session"}}),
+            ]
+        )
+        + "\n",
+    )
+
+    assert result.returncode == 0, result.stderr
+    frames = _stdout_frames(result)
+    assert [frame["type"] for frame in frames] == ["control_response", "control_response"]
+    assert frames[0]["response"]["subtype"] == "success"
+    assert frames[0]["response"]["request_id"] == "req-init"
+    assert "interrupt" in frames[0]["response"]["response"]["capabilities"]
+    assert frames[1]["response"]["subtype"] == "success"
+    assert frames[1]["response"]["request_id"] == "req-end"
+
+
+def test_process_subprocess_rejects_invalid_json(tmp_path) -> None:
+    result = _run_process_cli(config_dir=tmp_path, stdin="{not-json\n")
+
+    assert result.returncode == 1
+    frames = _stdout_frames(result)
+    assert frames == [
+        {
+            "type": "error",
+            "request_id": None,
+            "error": {"code": "invalid_json", "message": "Invalid JSON frame.", "retryable": False},
+        }
+    ]
+
+
+def test_process_subprocess_pipeline_mode_initialize_and_end_session(tmp_path) -> None:
+    result = _run_process_cli(
+        config_dir=tmp_path,
+        mode="pipeline",
+        stdin="\n".join(
+            [
+                json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}}),
+                json.dumps({"type": "control_request", "request_id": "req-end", "request": {"subtype": "end_session"}}),
+            ]
+        )
+        + "\n",
+    )
+
+    assert result.returncode == 0, result.stderr
+    frames = _stdout_frames(result)
+    assert [frame["type"] for frame in frames] == ["control_response", "control_response"]
+    assert frames[0]["response"]["subtype"] == "success"
+    assert "pipeline" in frames[0]["response"]["response"]["capabilities"]
+    assert "pipeline_resume" in frames[0]["response"]["response"]["capabilities"]
+    assert frames[1]["response"]["subtype"] == "success"
+
+
+def test_process_subprocess_pipeline_mode_returns_recoverable_task_id(tmp_path) -> None:
+    context_dir = tmp_path / "process-pipeline" / "contexts"
+    context_dir.mkdir(parents=True)
+    (context_dir / "ctx-1.json").write_text(
+        json.dumps(
+            {
+                "contextId": "ctx-1",
+                "taskId": "task-1",
+                "iacCodeSessionId": "iac-session-1",
+                "cwd": str(REPO_ROOT),
+                "sidecarStatus": "waiting_input",
+                "activeTaskId": "task-1",
+            }
+        ),
+        encoding="utf-8",
+    )
+    result = _run_process_cli(
+        config_dir=tmp_path,
+        mode="pipeline",
+        stdin="\n".join(
+            [
+                json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}}),
+                json.dumps(
+                    {
+                        "type": "user",
+                        "request_id": "req-resume",
+                        "metadata": {"iac_code": {"contextId": "ctx-1"}},
+                        "message": {"role": "user", "content": "continue"},
+                    }
+                ),
+            ]
+        )
+        + "\n",
+    )
+
+    assert result.returncode == 0, result.stderr
+    frames = _stdout_frames(result)
+    error = next(frame for frame in frames if frame["type"] == "error")
+    assert error["request_id"] == "req-resume"
+    assert error["error"]["code"] == "pipeline_task_required"
+    assert error["error"]["data"] == {
+        "contextId": "ctx-1",
+        "recoverableTaskId": "task-1",
+        "sidecarStatus": "waiting_input",
+    }
+
+
+def test_process_subprocess_unknown_mode_uses_normal_fallback(tmp_path) -> None:
+    result = _run_process_cli(
+        config_dir=tmp_path,
+        mode="unexpected",
+        stdin=json.dumps({"type": "control_request", "request_id": "req-init", "request": {"subtype": "initialize"}})
+        + "\n",
+    )
+
+    assert result.returncode == 0, result.stderr
+    frames = _stdout_frames(result)
+    assert len(frames) == 1
+    assert frames[0]["type"] == "control_response"
+    assert frames[0]["response"]["subtype"] == "success"
diff --git a/tests/services/permissions/test_trusted_roots.py b/tests/services/permissions/test_trusted_roots.py
index 6e17a936..0cf642fb 100644
--- a/tests/services/permissions/test_trusted_roots.py
+++ b/tests/services/permissions/test_trusted_roots.py
@@ -1,5 +1,6 @@
 import importlib
 from pathlib import Path
+from unittest.mock import MagicMock
 
 import pytest
 
@@ -34,6 +35,21 @@ def test_build_session_trusted_read_directories_includes_session_scoped_artifact
     ]
 
 
+def test_build_session_trusted_read_directories_ignores_mock_session_dir(monkeypatch, tmp_path):
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr("iac_code.config.get_config_dir", lambda: tmp_path / ".iac-code")
+
+    from iac_code.services.permissions.trusted_roots import build_session_trusted_read_directories
+
+    roots = build_session_trusted_read_directories("abc123", session_dir=MagicMock())
+
+    assert roots == [
+        str(Path(tmp_path / ".iac-code" / "tool-results" / "abc123")),
+        str(Path(tmp_path / ".iac-code" / "image-cache" / "abc123")),
+    ]
+    assert not (tmp_path / "MagicMock").exists()
+
+
 def test_build_session_trusted_read_directories_rejects_symlinked_session_artifact_dir(monkeypatch, tmp_path):
     monkeypatch.setattr("iac_code.config.get_config_dir", lambda: tmp_path / ".iac-code")
 
diff --git a/tests/ui/test_repl_integration.py b/tests/ui/test_repl_integration.py
index 68a161f9..3d851191 100644
--- a/tests/ui/test_repl_integration.py
+++ b/tests/ui/test_repl_integration.py
@@ -8,7 +8,7 @@
 import sys
 from pathlib import Path
 from types import SimpleNamespace
-from unittest.mock import AsyncMock, Mock, patch
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
 
@@ -468,6 +468,20 @@ def test_session_dir_for_artifacts_returns_none_for_legacy_directory_without_lay
     assert repl._result_storage_dir_for_session() is None
 
 
+def test_session_dir_for_artifacts_ignores_mock_storage_paths(tmp_path, monkeypatch):
+    from iac_code.ui.repl import InlineREPL
+
+    monkeypatch.chdir(tmp_path)
+    repl = InlineREPL.__new__(InlineREPL)
+    repl._session_storage = MagicMock()
+    repl._original_cwd = str(tmp_path)
+    repl._session_id = "mock-session"
+
+    assert repl._session_dir_for_artifacts() is None
+    assert repl._raw_session_dir_for_trusted_roots() is None
+    assert not (tmp_path / "MagicMock").exists()
+
+
 def test_history_search_uses_agent_context_messages():
     from iac_code.agent.message import Message
     from iac_code.state.app_state import AppState
diff --git a/website/docs/automation/non-interactive-mode.md b/website/docs/automation/non-interactive-mode.md
index 7be4556e..f11e80c4 100644
--- a/website/docs/automation/non-interactive-mode.md
+++ b/website/docs/automation/non-interactive-mode.md
@@ -39,6 +39,28 @@ Supported output formats are:
 | `json` | A single JSON result for callers that parse the final response. |
 | `stream-json` | Streaming JSON events for callers that process incremental progress. |
 
+## SDK Process Mode
+
+SDK process mode is for clients that keep `iac-code` running as a subprocess and exchange line-delimited JSON over stdin/stdout:
+
+```bash
+iac-code --input-format stream-json --output-format stream-json
+```
+
+This is separate from one-shot `--prompt` mode. `--input-format stream-json` requires `--output-format stream-json` and rejects `--prompt`. The caller must send an `initialize` control request before user messages:
+
+```json
+{"type":"control_request","request_id":"req-init","request":{"subtype":"initialize","cwd":"/absolute/workspace","model":"qwen3.7-max"}}
+{"type":"user","request_id":"req-1","session_id":"session-1","message":{"role":"user","content":"Create an OSS Bucket"},"metadata":{"iac_code":{"cwd":"/absolute/workspace"}}}
+{"type":"control_request","request_id":"req-end","request":{"subtype":"end_session"}}
+```
+
+The process writes `control_response`, `stream_event`, `error`, and final `result` frames. Stream events reuse the public `stream-json` event shape used by one-shot output streaming. `cwd` values in initialize payloads or `metadata.iac_code.cwd` must be absolute paths.
+
+Supported control flows include `initialize`, `interrupt`, `set_model`, `end_session`, `close`, `keep_alive`, and `update_environment_variables`. Only one user turn can run at a time inside a process. If two subprocesses try to use the same `session_id` concurrently for the same `cwd`, the second turn receives a retryable `session_busy` error.
+
+When `IAC_CODE_MODE=pipeline`, process mode executes Pipeline mode instead of the normal agent loop. Pipeline stream frames use `type: "pipeline_event"`, and final result frames include a `pipeline` object with `contextId`, `taskId`, `iacCodeSessionId`, `status`, and `sidecarStatus`. For a recoverable pipeline follow-up, send the same `contextId` and the active `taskId`; if the context has a recoverable task and the task id is omitted, the process returns a retryable `pipeline_task_required` error with `recoverableTaskId`.
+
 ## Session Backups
 
 When `IAC_CODE_CONFIG_BACKUP_DIR` is set, non-interactive runs mirror the v2 session at key checkpoints. The ordinary end-of-turn checkpoint uses `normal_turn_end`; backup failures at that point are logged as a `warning` and recorded in `.backup-state.json` without failing the completed response or adding a warning field to the final output. Pipeline mode has its own critical backup gates.
diff --git a/website/docs/automation/pipeline-mode.md b/website/docs/automation/pipeline-mode.md
index f329de15..8d728430 100644
--- a/website/docs/automation/pipeline-mode.md
+++ b/website/docs/automation/pipeline-mode.md
@@ -21,7 +21,7 @@ Design a low-cost Alibaba Cloud web application deployment and generate a templa
 
 ## Start Pipeline Mode
 
-Pipeline mode currently requires the interactive REPL. It cannot be combined with `--prompt`.
+Pipeline mode can run through the interactive REPL or through SDK process mode. It cannot be combined with `--prompt`.
 
 On macOS or Linux:
 
@@ -42,6 +42,12 @@ The default pipeline name is `selling`. To be explicit:
 IAC_CODE_MODE=pipeline IAC_CODE_PIPELINE_NAME=selling iac-code
 ```
 
+For SDK subprocess clients, start process mode with stream-json input and output:
+
+```bash
+IAC_CODE_MODE=pipeline iac-code --input-format stream-json --output-format stream-json
+```
+
 ## Pipeline and selling
 
 | Name | Meaning |
@@ -89,14 +95,16 @@ After the pipeline completes, fails, exits early, or is canceled, IaC Code switc
 
 ## Automation Integrations
 
-Pipeline mode is currently primarily designed for the interactive REPL. A2A server mode can expose pipeline progress, artifacts, permission results, and recovery information, which is useful when connecting a pipeline to an external console or task system.
+Pipeline mode can be integrated through A2A server mode or SDK process mode. A2A server mode exposes pipeline progress, artifacts, permission results, and recovery information for external consoles or task systems. SDK process mode keeps `iac-code` as a local subprocess and exchanges line-delimited JSON over stdin/stdout.
+
+In SDK process mode, pipeline events are emitted as `stream_event` frames whose event payload has `type: "pipeline_event"`. Final `result` frames include a `pipeline` object with `contextId`, `taskId`, `iacCodeSessionId`, `status`, and `sidecarStatus`. A paused pipeline should be resumed by sending the same `contextId` and active `taskId`. If a context has a recoverable task and the client omits the task id, the process returns a retryable `pipeline_task_required` error with `recoverableTaskId`.
 
 ACP does not currently support Pipeline mode. `--prompt` / [Non-interactive Mode](./non-interactive-mode.md) runs a normal one-shot request and does not execute Pipeline steps.
 
 ## Current Limitations
 
 - The current release includes only the `selling` pipeline, mainly for Alibaba Cloud infrastructure workflows.
-- Pipeline mode requires the interactive REPL. `--prompt` is rejected when `IAC_CODE_MODE=pipeline`.
+- Pipeline mode supports the interactive REPL and SDK process mode. `--prompt` is rejected when `IAC_CODE_MODE=pipeline`.
 - Pipeline mode supports text input. Images pasted into the REPL are ignored while the pipeline is active.
 - Mid-pipeline shell escapes, skill triggers, and most slash commands are restricted unless the pipeline definition explicitly allows them. Basic commands such as `/help`, `/status`, `/resume`, and `/exit` remain available.
 
diff --git a/website/docs/cli/command-line-options.md b/website/docs/cli/command-line-options.md
index 66e7748b..9b7f38b8 100644
--- a/website/docs/cli/command-line-options.md
+++ b/website/docs/cli/command-line-options.md
@@ -13,7 +13,8 @@ Command line options change how IaC Code starts. Use them before entering the in
 | `-v`, `-V`, `--version` | Print the installed IaC Code version and exit. |
 | `-m <model>`, `--model <model>` | Start with a specific LLM model. This overrides the saved model for the current run. |
 | `-p <prompt>`, `--prompt <prompt>` | Run a single prompt and exit. This enables non-interactive mode. Use `--prompt -` to read the prompt from standard input. |
-| `--output-format <format>` | Set output format for non-interactive mode. Supported values are `text`, `json`, and `stream-json`. The default is `text`. |
+| `--output-format <format>` | Set output format for one-shot or process automation. Supported values are `text`, `json`, and `stream-json`. The default is `text`. |
+| `--input-format <format>` | Set input format. Supported values are `text` and `stream-json`. `stream-json` starts SDK process mode and must be combined with `--output-format stream-json`; it cannot be combined with `--prompt`. |
 | `--max-turns <number>` | Limit the maximum number of agent turns in non-interactive mode. The default is `100`. |
 | `--thinking-enabled`, `--no-thinking-enabled` | Control whether one-shot non-interactive requests explicitly enable thinking. The default is `--thinking-enabled`; use `--no-thinking-enabled` to send `thinking_enabled=false` for this run without rewriting `settings.yml`. |
 | `-d`, `--debug` | Enable debug logging for the current run. In interactive mode, use `/debug` to inspect or change debug logging after startup. |
@@ -91,3 +92,11 @@ Run in automation with no interactive prompts:
 ```bash
 iac-code --prompt "Create a VPC" --permission-mode bypass_permissions
 ```
+
+Start SDK process mode for a long-running subprocess client:
+
+```bash
+iac-code --input-format stream-json --output-format stream-json
+```
+
+In process mode the caller writes one JSON frame per line to stdin and reads one JSON frame per line from stdout. Send an `initialize` control request before user messages. This mode supports `interrupt`, `set_model`, `end_session`, `close`, `keep_alive`, and `update_environment_variables` control flows. When `IAC_CODE_MODE=pipeline`, the same process entry point runs Pipeline mode and returns pipeline status in stream and result frames.
diff --git a/website/docs/configuration/runtime-configuration.md b/website/docs/configuration/runtime-configuration.md
index efa6e586..6c21484b 100644
--- a/website/docs/configuration/runtime-configuration.md
+++ b/website/docs/configuration/runtime-configuration.md
@@ -206,3 +206,12 @@ Key v2 session paths are:
 <config-dir>/projects/<project>/<session-id>/pipeline/transcripts/<transcript-id>/permission-audit.jsonl
 <config-dir>/projects/<project>/<session-id>/pipeline/transcripts/<transcript-id>/tool-results/
 ```
+
+SDK process mode also stores cross-process Pipeline recovery snapshots under:
+
+```text
+<config-dir>/process-pipeline/contexts/<encoded-context-id>.json
+<config-dir>/process-pipeline/contexts/<encoded-context-id>.lock
+```
+
+Each snapshot records `contextId`, `taskId`, `iacCodeSessionId`, `cwd`, `sidecarStatus`, and `activeTaskId`. These files let a later SDK process recover which task belongs to a Pipeline context and return `pipeline_task_required` with a `recoverableTaskId` when the caller omits the active task id. They do not replace the session-owned Pipeline state above; they are a small routing and recovery index, protected with per-context file locks for multi-process clients. A process-mode Pipeline follow-up is rejected if the stored context belongs to a different `cwd`.
diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/automation/non-interactive-mode.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/automation/non-interactive-mode.md
index d929430c..9d6a975f 100644
--- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/automation/non-interactive-mode.md
+++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/automation/non-interactive-mode.md
@@ -39,6 +39,28 @@ iac-code --prompt "创建一个 VPC" --max-turns 20
 | `json` | 返回单个 JSON 结果，适合调用方解析最终响应。 |
 | `stream-json` | 输出流式 JSON 事件，适合调用方处理增量进度。 |
 
+## SDK Process 模式
+
+SDK process 模式用于让客户端把 `iac-code` 作为长期运行的子进程，并通过 stdin/stdout 交换按行分隔的 JSON：
+
+```bash
+iac-code --input-format stream-json --output-format stream-json
+```
+
+这不同于一次性的 `--prompt` 模式。`--input-format stream-json` 必须与 `--output-format stream-json` 一起使用，并且会拒绝 `--prompt`。调用方必须先发送 `initialize` control request，然后才能发送用户消息：
+
+```json
+{"type":"control_request","request_id":"req-init","request":{"subtype":"initialize","cwd":"/absolute/workspace","model":"qwen3.7-max"}}
+{"type":"user","request_id":"req-1","session_id":"session-1","message":{"role":"user","content":"创建一个 OSS Bucket"},"metadata":{"iac_code":{"cwd":"/absolute/workspace"}}}
+{"type":"control_request","request_id":"req-end","request":{"subtype":"end_session"}}
+```
+
+进程会输出 `control_response`、`stream_event`、`error` 和最终 `result` frame。流式事件复用一次性输出 streaming 使用的公开 `stream-json` event 结构。initialize payload 或 `metadata.iac_code.cwd` 中的 `cwd` 必须是绝对路径。
+
+支持的控制流程包括 `initialize`、`interrupt`、`set_model`、`end_session`、`close`、`keep_alive` 和 `update_environment_variables`。同一个 process 内一次只能运行一个用户 turn。如果两个子进程尝试在同一个 `cwd` 下并发使用同一个 `session_id`，第二个 turn 会收到可重试的 `session_busy` 错误。
+
+当 `IAC_CODE_MODE=pipeline` 时，process 模式会执行 Pipeline 模式，而不是普通 agent loop。Pipeline stream frame 使用 `type: "pipeline_event"`，最终 result frame 会包含 `pipeline` 对象，其中有 `contextId`、`taskId`、`iacCodeSessionId`、`status` 和 `sidecarStatus`。恢复可继续的 pipeline 时，需要发送同一个 `contextId` 和活跃的 `taskId`；如果 context 中存在可恢复 task 但调用方省略了 task id，process 会返回可重试的 `pipeline_task_required` 错误，并携带 `recoverableTaskId`。
+
 ## 会话备份
 
 设置 `IAC_CODE_CONFIG_BACKUP_DIR` 后，非交互运行会在关键检查点镜像 v2 session。普通轮次结束检查点使用 `normal_turn_end`；此时备份失败只会记录为 `warning` 并写入 `.backup-state.json`，不会让已完成响应失败，也不会在最终输出中新增 warning 字段。Pipeline 模式有自己的关键备份 gate。
diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/automation/pipeline-mode.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/automation/pipeline-mode.md
index 57c2b851..bc3fbda7 100644
--- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/automation/pipeline-mode.md
+++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/automation/pipeline-mode.md
@@ -21,7 +21,7 @@ Pipeline 本身是通用能力；当前内置实现的是 `selling` pipeline。`
 
 ## 启动 Pipeline 模式
 
-Pipeline 模式目前需要交互式 REPL，不能和 `--prompt` 一起使用。
+Pipeline 模式可以通过交互式 REPL 或 SDK process 模式运行，不能和 `--prompt` 一起使用。
 
 在 macOS 或 Linux 上：
 
@@ -42,6 +42,12 @@ iac-code
 IAC_CODE_MODE=pipeline IAC_CODE_PIPELINE_NAME=selling iac-code
 ```
 
+对于 SDK 子进程客户端，可以用 stream-json 输入和输出启动 process 模式：
+
+```bash
+IAC_CODE_MODE=pipeline iac-code --input-format stream-json --output-format stream-json
+```
+
 ## Pipeline 与 selling 的关系
 
 | 名称 | 含义 |
@@ -89,14 +95,16 @@ Pipeline 运行时可能会暂停等待用户输入，例如：
 
 ## 自动化集成
 
-Pipeline 模式目前主要面向交互式 REPL。A2A 服务模式可以对外暴露 pipeline 进度、产物、权限结果和恢复信息，适合把 pipeline 接入外部控制台或任务系统。
+Pipeline 模式可以通过 A2A 服务模式或 SDK process 模式集成。A2A 服务模式会对外暴露 pipeline 进度、产物、权限结果和恢复信息，适合把 pipeline 接入外部控制台或任务系统。SDK process 模式会把 `iac-code` 保持为本地子进程，并通过 stdin/stdout 交换按行分隔的 JSON。
+
+在 SDK process 模式中，pipeline 事件会以 `stream_event` frame 发出，事件 payload 的 `type` 为 `"pipeline_event"`。最终 `result` frame 会包含 `pipeline` 对象，其中有 `contextId`、`taskId`、`iacCodeSessionId`、`status` 和 `sidecarStatus`。暂停中的 pipeline 应使用相同的 `contextId` 和活跃 `taskId` 恢复。如果某个 context 存在可恢复 task 但客户端省略 task id，process 会返回可重试的 `pipeline_task_required` 错误，并携带 `recoverableTaskId`。
 
 ACP 目前不支持 Pipeline 模式。`--prompt` / [非交互模式](./non-interactive-mode.md) 会走普通一次性调用，不会执行 Pipeline 步骤。
 
 ## 当前限制
 
 - 当前内置 pipeline 只有 `selling`，主要面向阿里云基础设施工作流。
-- Pipeline 模式需要交互式 REPL；当 `IAC_CODE_MODE=pipeline` 时，`--prompt` 会被拒绝。
+- Pipeline 模式支持交互式 REPL 和 SDK process 模式；当 `IAC_CODE_MODE=pipeline` 时，`--prompt` 会被拒绝。
 - Pipeline 模式支持文本输入。Pipeline 激活时，粘贴到 REPL 的图片会被忽略。
 - Pipeline 运行期间，shell escape、技能触发器和大多数 slash command 会被限制，除非 pipeline 定义显式允许。`/help`、`/status`、`/resume`、`/exit` 等基础命令仍然可用。
 
diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/cli/command-line-options.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/cli/command-line-options.md
index 0de4be01..2080fc67 100644
--- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/cli/command-line-options.md
+++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/cli/command-line-options.md
@@ -13,7 +13,8 @@ description: IaC Code 启动选项和一次性执行参数参考。
 | `-v`, `-V`, `--version` | 输出已安装的 IaC Code 版本并退出。 |
 | `-m <model>`, `--model <model>` | 使用指定的 LLM 模型启动。本次运行会覆盖已保存的模型设置。 |
 | `-p <prompt>`, `--prompt <prompt>` | 执行单条提示词并退出。这会进入非交互模式。使用 `--prompt -` 可以从标准输入读取提示词。 |
-| `--output-format <format>` | 设置非交互模式的输出格式。支持 `text`、`json` 和 `stream-json`，默认值为 `text`。 |
+| `--output-format <format>` | 设置一次性调用或 process 自动化的输出格式。支持 `text`、`json` 和 `stream-json`，默认值为 `text`。 |
+| `--input-format <format>` | 设置输入格式。支持 `text` 和 `stream-json`。`stream-json` 会启动 SDK process 模式，必须与 `--output-format stream-json` 一起使用，且不能与 `--prompt` 组合。 |
 | `--max-turns <number>` | 限制非交互模式中的最大代理轮次，默认值为 `100`。 |
 | `--thinking-enabled`, `--no-thinking-enabled` | 控制一次性非交互请求是否显式启用 thinking。默认值为 `--thinking-enabled`；使用 `--no-thinking-enabled` 会在本次运行中发送 `thinking_enabled=false`，但不会改写 `settings.yml`。 |
 | `-d`, `--debug` | 为本次运行启用调试日志。交互模式启动后，可以使用 `/debug` 查看或调整调试日志。 |
@@ -91,3 +92,11 @@ iac-code --allowed-tools 'aliyun_api(ros:CreateStack)'
 ```bash
 iac-code --prompt "创建一个 VPC" --permission-mode bypass_permissions
 ```
+
+为长期运行的子进程客户端启动 SDK process 模式：
+
+```bash
+iac-code --input-format stream-json --output-format stream-json
+```
+
+在 process 模式下，调用方向 stdin 写入一行一个 JSON frame，并从 stdout 读取一行一个 JSON frame。发送用户消息前需要先发送 `initialize` control request。该模式支持 `interrupt`、`set_model`、`end_session`、`close`、`keep_alive` 和 `update_environment_variables` 等控制流程。当 `IAC_CODE_MODE=pipeline` 时，同一个 process 入口会运行 Pipeline 模式，并在 stream 和 result frame 中返回 pipeline 状态。
diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/configuration/runtime-configuration.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/configuration/runtime-configuration.md
index 6cb12a18..fa978e3a 100644
--- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/configuration/runtime-configuration.md
+++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/configuration/runtime-configuration.md
@@ -206,3 +206,12 @@ v2 session 的关键路径如下：
 <config-dir>/projects/<project>/<session-id>/pipeline/transcripts/<transcript-id>/permission-audit.jsonl
 <config-dir>/projects/<project>/<session-id>/pipeline/transcripts/<transcript-id>/tool-results/
 ```
+
+SDK process 模式还会把跨进程 Pipeline 恢复快照存放在：
+
+```text
+<config-dir>/process-pipeline/contexts/<encoded-context-id>.json
+<config-dir>/process-pipeline/contexts/<encoded-context-id>.lock
+```
+
+每个快照会记录 `contextId`、`taskId`、`iacCodeSessionId`、`cwd`、`sidecarStatus` 和 `activeTaskId`。这些文件让后续 SDK process 能恢复某个 Pipeline context 对应的 task，并在调用方省略活跃 task id 时返回带有 `recoverableTaskId` 的 `pipeline_task_required`。它们不会替代上面的 session 内 Pipeline 状态，只是一个小型路由和恢复索引；为支持多进程客户端，每个 context 都会使用独立文件锁保护。process 模式的 Pipeline 后续请求如果发现已保存 context 属于不同 `cwd`，会被拒绝。