"""Bundled Grok provider catalog.""" import json from pathlib import Path from typing import Any, Mapping from urllib.parse import quote from okstra_ctl.domain.provider import ( LeadLaunchSpec, ModelSpec, ProviderSpec, ServedModelAttestation, ) from okstra_ctl.domain.role import ALL_ROLE_TOKENS from okstra_ctl.domain.worker_exec import ( ExecCommand, PolicySupport, WorkerExecRequest, ) from okstra_ctl.domain.worker_presentation import JsonEvents from okstra_ctl.domain.worker_stream import StreamEvent, Text, ToolCall, ToolResult GROK = { "grok-4.6": ModelSpec("grok-4.6", "grok-4.6", "grok-4.6", pricing=(2.00, 0.50, 6.00)), "grok-4.5": ModelSpec("grok-4.5", "grok-4.5", "grok-4.5", pricing=(2.00, 0.30, 6.00)), "grok-build-0.1": ModelSpec("grok-build-0.1", "grok-build-0.1", "grok-build-0.1", pricing=(1.00, 0.20, 2.00)), } def _catalog_model_id(observed: str) -> str: """Undo the `-build` suffix grok reports its served model under. Measured on grok 1.x: asking for `grok-4.6` closes with `modelUsage: {"grok-4.6-build": ...}`, and `grok-4.5` with `grok-4.5-build`. The catalog holds the requested ids, so reading the reported one as-is would fail the served-model gate on the very model okstra asked for. A catalog entry that already carries the suffix (`grok-build-0.1`) is matched first and left alone. """ if observed in GROK: return observed bare = observed.removesuffix("-build") return bare if bare != observed and bare in GROK else observed def normalise_served_model(raw_model: str | None) -> ServedModelAttestation: if not raw_model or not raw_model.strip(): return ServedModelAttestation.unknown() model_id = raw_model.strip().lower().removeprefix("grok/") return ServedModelAttestation( raw_model, f"grok/{_catalog_model_id(model_id)}", "exact", "provider-output", ) def served_model_from_session( session_id: str, cwd: Path, *, home: Path | None = None ) -> str | None: """세션 기록에서 실제로 서빙된 모델을 읽는다. grok 은 일하는 동안 어느 이벤트에도 모델을 적지 않고, 닫는 기록에만 남긴다. 스트림을 해석하지 않는 경로에서는 여기가 유일한 관측 지점이다. 디렉터리 이름은 cwd 를 퍼센트 인코딩한 것이다. """ root = (home or Path.home() / ".grok") / "sessions" encoded = quote(str(cwd), safe="") history = root / encoded / session_id / "chat_history.jsonl" if not history.is_file(): return None try: for line in history.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip(): continue record = json.loads(line) model = record.get("model_id") if isinstance(record, Mapping) else None if isinstance(model, str) and model.strip(): return model except (OSError, ValueError): return None return None def observe_served_model(event: Mapping[str, Any]) -> str | None: """Read the model grok actually served from its closing `end` event. grok names no model on the events it streams as it works; the only place it states one is `end.modelUsage`, keyed by model. Looking for a top-level `model` field — Claude's shape — found nothing on every event, so the run recorded `servedModelAttestation: unknown` and the served-model gate had nothing to check. """ model = event.get("model") if isinstance(model, str) and model.strip(): return model message = event.get("message") if isinstance(message, Mapping): model = message.get("model") if isinstance(model, str) and model.strip(): return model if event.get("type") == "end": usage = event.get("modelUsage") if isinstance(usage, Mapping): names = [name for name in usage if isinstance(name, str) and name.strip()] # More than one model in a single turn means the CLI switched # mid-run; naming either one would be a guess, so report neither. if len(names) == 1: return names[0] return None def stream_events(event: Mapping[str, Any]) -> tuple[StreamEvent, ...]: """grok `--output-format streaming-json` 한 줄을 공통 이벤트로 옮긴다. 이 CLI 의 `-p` 기본 `plain` 은 종료 시에만 stdout 에 쓴다. 파이프에 붙은 워커는 그 동안 pane 과 로그가 0바이트다(실측 2026-08-22). 같은 호출에 `streaming-json` 을 주면 ACP 세션 업데이트가 NDJSON 으로 흐른다. 키는 `type=text|tool_call|tool_call_update|end` 이고 Claude 의 `message.content` 가 아니다 — 그 스키마로 읽으면 이벤트는 전부 버려진다. """ kind = event.get("type") if kind == "text": data = event.get("data") # 공백만 있는 토큰도 살린다. 버리면 붙인 문장에서 단어가 붙는다. return (Text(body=data),) if isinstance(data, str) and data else () if kind == "tool_call": name = str(event.get("toolName") or event.get("title") or "tool") return (ToolCall(name=name, detail=_call_detail(event)),) if kind == "tool_call_update": return _completed_tool(event) if kind == "error": message = event.get("message") return (Text(body=message),) if isinstance(message, str) and message.strip() else () return () def _call_detail(event: Mapping[str, Any]) -> str: payload = event.get("rawInput") if isinstance(payload, Mapping): for key in ("command", "file_path", "path", "pattern", "query"): value = payload.get(key) if value: return str(value) return next( (str(value) for value in payload.values() if isinstance(value, str) and value), "", ) title = event.get("title") return str(title) if isinstance(title, str) else "" def _completed_tool(event: Mapping[str, Any]) -> tuple[StreamEvent, ...]: if event.get("status") != "completed": return () body = _output_body(event.get("rawOutput")) return (ToolResult(body=body, size_bytes=len(body.encode("utf-8"))),) def _output_body(output: Any) -> str: if isinstance(output, str): return output if isinstance(output, Mapping): text = next( (str(value) for value in output.values() if isinstance(value, str) and value), "", ) return text or json.dumps(output, ensure_ascii=False) return "" if output is None else str(output) GROK_LEAD_LAUNCH = LeadLaunchSpec( executable="grok", model_flag="--model", resume_session_id_flag="--resume", ) class GrokExecution: """grok CLI invocation. Runs inside the stage tree when there is one. That is where this provider's production path already puts it, and it is why the working directory is part of the returned command rather than something the runner picks. """ def build_command(self, request: WorkerExecRequest) -> ExecCommand: cwd = request.worktree_path or request.project_root # `-p` 기본 `plain` 은 종료 시에만 stdout 에 쓴다. 러너는 파이프라 # TUI 도 열리지 않고, `--no-alt-screen` 도 그 침묵을 바꾸지 않는다. return ExecCommand( argv=( "grok", "-p", request.prompt_text, "-m", request.model, "--cwd", str(cwd), "--output-format", "streaming-json", ), stdin_text=None, cwd=cwd, presentation=JsonEvents( normalise=stream_events, observe=observe_served_model, join_adjacent_text=True, ), ) def policy_support(self) -> PolicySupport: return PolicySupport( can_auto_approve=False, can_bound_write_scope=False, note="this CLI exposes no approval or sandbox flag", ) def create_provider() -> ProviderSpec: return ProviderSpec( provider="grok", display_label="Grok", models=GROK, default_models={ role: "grok-4.6" for role in ( "lead", "analyser", "critic", "designer", "planner", "executor", "verifier", "report-writer", "translator", ) }, wrapper="okstra-grok-exec.sh", supported_roles=ALL_ROLE_TOKENS, execution_capabilities=frozenset( { "lead-session", "worker-artifact-io", "source-readonly", "extended-artifact-authoring", "project-mutation", } ), lead_launch=GROK_LEAD_LAUNCH, exec_strategy=GrokExecution(), served_model_normalizer=normalise_served_model, )