[feat] agent: record token usage, compact project views, and precheck

apps/api/src/opendirector_api/creation.py

@@ -179,13 +179,20 @@ def projection(project_id: UUID):
         return canvas.projection(project_id)
 
     @router.post('/projects/{project_id}/queries')
-    async def query(project_id: UUID, value: CreationQuery):
+    async def query(project_id: UUID, value: CreationQuery, request: Request):
         if value.projectId != project_id:
             raise ProjectError('查询不属于当前工程。')
         result = await canvas.query(value)
         if value.type == 'asset.preview' and 'preview' in result:
             result['preview'].pop('path')
             result['preview']['url'] = f'/api/creation/projects/{project_id}/assets/{value.objectId}/preview'
+        if value.type == 'generation.precheck' and isinstance(result, dict):
+            language = request.headers.get('accept-language')
+            for item in result.get('items', ()):
+                for issue in item.get('issues', ()):
+                    issue['message'] = localized_error_for_language(
+                        language, chinese=issue['message'], status=422, code=issue['code'],
+                    )
         return result
 
     @router.get('/projects/{project_id}/assets/{asset_id}/preview')

docs/guides/cli.md

@@ -42,6 +42,25 @@ CLI 输出 JSON。修改回执附带对应的 assetId、itemId、specId 及其
 Agent 的 Shell 通过 PATH 使用同一个 `opendirector`。宿主提供受令牌保护的本轮连接、工程范围、编辑会话和执行目录，文件登记限制在本轮执行目录。每次调用产生业务活动及回执，供聊天与轨迹页展示。
 
 
+## 紧凑工程索引与生成前预检
+
+多节点工程中，先用紧凑摘要定位目标画布项、对象版本与活动任务，再按 `--item-id` 读取局部对象树，避免在每轮 Agent 上下文中反复装载完整历史与长提示词。批量或长时长提交前，通过 `generation precheck` 执行无副作用预检（不创建任务、不增加版本），提前校验模型协议、锁定状态、排队上限、素材可用性与预计总时长。
+
+```bash
+opendirector --project PROJECT_UUID project show --view summary
+opendirector --project PROJECT_UUID project show --item-id ITEM_UUID
+opendirector --project PROJECT_UUID generation precheck --spec-id SPEC_UUID_1 --spec-id SPEC_UUID_2
+```
+
+`generation precheck` 全部通过时退出码为 0，任一配置存在阻断问题时输出完整报告并返回退出码 1。Agent 工具 `read_project`（支持 `view` 与 `itemIds`）和 `query`（`generation.precheck`）共用同一后端查询契约。
+
+### Compact project views and generation precheck
+
+On multi-node projects, inspect `--view summary` first to locate canvas items, object revisions, and active jobs, then scope reads with `--item-id` so Agent turns do not repeatedly load unrelated long prompts and historical results. Before submitting batch or long-duration generations, run `generation precheck` to validate model rules, locks, queue limits, reference byte budgets, and estimated video seconds without creating jobs or commits.
+
+`generation precheck` exits with code 0 when every configuration is ready, or prints the diagnostic report and exits with code 1 when any configuration is blocked. The Agent tools `read_project` (`view`, `itemIds`) and `query` (`generation.precheck`) call the same backend query service.
+
+
 ## 检查已有生成结果
 
 `job show` / `job wait` 在任务成功后返回 `delivery`：已登记的 AssetRef、当前采用状态及展示该素材的画布项。生成成功不等于质量验收；Agent 应检查现有素材。复制到执行目录做媒体探测、抽帧和审看不需要重新发布，也不应为同一次交付新增画布项。编辑产生的新交付文件才需要登记；独立副本展示应有用户要求。

packages/core/src/opendirector/agent/model_gateway.py

@@ -20,6 +20,94 @@
 )
 
 logger = logging.getLogger(__name__)
+MAX_JSON_USAGE_BYTES = 2 * 1024 * 1024
+
+
+def _token_count(value: object) -> int | None:
+    if isinstance(value, int) and not isinstance(value, bool) and value >= 0:
+        return value
+    return None
+
+
+def normalize_usage(raw: object) -> dict[str, int] | None:
+    """Extract provider-neutral token and KV-cache counters without trusting wire types."""
+    if not isinstance(raw, dict):
+        return None
+    prompt = _token_count(raw.get("prompt_tokens"))
+    if prompt is None:
+        prompt = _token_count(raw.get("input_tokens"))
+    completion = _token_count(raw.get("completion_tokens"))
+    if completion is None:
+        completion = _token_count(raw.get("output_tokens"))
+    total = _token_count(raw.get("total_tokens"))
+    if total is None and prompt is not None and completion is not None:
+        total = prompt + completion
+
+    prompt_details = raw.get("prompt_tokens_details")
+    if not isinstance(prompt_details, dict):
+        prompt_details = raw.get("input_tokens_details")
+    if not isinstance(prompt_details, dict):
+        prompt_details = {}
+    completion_details = raw.get("completion_tokens_details")
+    if not isinstance(completion_details, dict):
+        completion_details = raw.get("output_tokens_details")
+    if not isinstance(completion_details, dict):
+        completion_details = {}
+
+    cached = _token_count(prompt_details.get("cached_tokens"))
+    if cached is None:
+        cached = _token_count(raw.get("prompt_cache_hit_tokens"))
+    if cached is None:
+        cached = _token_count(raw.get("cache_read_input_tokens"))
+    cache_write = _token_count(raw.get("cache_creation_input_tokens"))
+    cache_miss = _token_count(raw.get("prompt_cache_miss_tokens"))
+    if cache_miss is None and prompt is not None and cached is not None:
+        cache_miss = max(0, prompt - cached)
+    reasoning = _token_count(completion_details.get("reasoning_tokens"))
+    if reasoning is None:
+        reasoning = _token_count(raw.get("reasoning_tokens"))
+
+    if (
+        prompt is None
+        and completion is None
+        and total is None
+        and cached is None
+        and cache_write is None
+    ):
+        return None
+
+    usage: dict[str, int] = {}
+    if prompt is not None:
+        usage["promptTokens"] = prompt
+    if completion is not None:
+        usage["completionTokens"] = completion
+    if total is not None:
+        usage["totalTokens"] = total
+    if cached is not None:
+        usage["cachedPromptTokens"] = cached
+    if cache_write is not None:
+        usage["cacheWritePromptTokens"] = cache_write
+    if cache_miss is not None:
+        usage["cacheMissPromptTokens"] = cache_miss
+    if reasoning is not None:
+        usage["reasoningTokens"] = reasoning
+    return usage
+
+
+def _merge_usage(previous: dict[str, int] | None, observed: dict[str, int] | None) -> dict[str, int] | None:
+    if observed is None:
+        return previous
+    if previous is None:
+        return dict(observed)
+    merged = {**previous, **observed}
+    prompt = merged.get("promptTokens")
+    completion = merged.get("completionTokens")
+    if prompt is not None and completion is not None and "totalTokens" not in observed:
+        merged["totalTokens"] = max(merged.get("totalTokens", 0), prompt + completion)
+    cached = merged.get("cachedPromptTokens")
+    if prompt is not None and cached is not None and "cacheMissPromptTokens" not in observed:
+        merged["cacheMissPromptTokens"] = max(0, prompt - cached)
+    return merged
 
 
 class ModelGateway:
@@ -33,6 +121,50 @@ def __init__(self, config: dict, thinking: str, token: str, project_id: str,
         self.trace_path = trace_path
         self.active: set[asyncio.Task] = set()
         self.closed = False
+        self._summary_written = False
+        self.request_count = 0
+        self.completed_requests = 0
+        self.usage_reported_requests = 0
+        self._cache_reported = False
+        self._totals = {
+            "promptTokens": 0,
+            "completionTokens": 0,
+            "totalTokens": 0,
+            "cachedPromptTokens": 0,
+            "cacheWritePromptTokens": 0,
+            "reasoningTokens": 0,
+        }
+
+    def usage_summary(self) -> dict[str, int | float | None]:
+        prompt = self._totals["promptTokens"]
+        cached = self._totals["cachedPromptTokens"]
+        hit_rate = round(min(cached, prompt) / prompt, 4) if self._cache_reported and prompt > 0 else None
+        return {
+            "requestCount": self.request_count,
+            "completedRequests": self.completed_requests,
+            "usageReportedRequests": self.usage_reported_requests,
+            **self._totals,
+            "cacheHitRate": hit_rate,
+        }
+
+    def close(self) -> dict[str, int | float | None]:
+        self.closed = True
+        summary = self.usage_summary()
+        if not self._summary_written and self.request_count > 0:
+            self._summary_written = True
+            self.trace("turn.usage", "summary", **summary)
+        return summary
+
+    def _record_completion(self, status_code: int, usage: dict[str, int] | None) -> None:
+        if 200 <= status_code < 300:
+            self.completed_requests += 1
+        if usage is None:
+            return
+        self.usage_reported_requests += 1
+        if "cachedPromptTokens" in usage:
+            self._cache_reported = True
+        for key in self._totals:
+            self._totals[key] += usage.get(key, 0)
 
     def trace(self, event: str, request_id: str, **details: object) -> None:
         if self.trace_path is None:
@@ -51,22 +183,22 @@ def trace(self, event: str, request_id: str, **details: object) -> None:
         except OSError:
             logger.exception("Cannot write model trace for turn %s", self.turn_id)
 
-    def trace_reasoning(self, request_id: str, data: str) -> None:
+    def trace_reasoning(self, request_id: str, data: str) -> dict[str, int] | None:
         try:
             event = json.loads(data)
         except json.JSONDecodeError:
-            return
+            return None
         if not isinstance(event, dict):
-            return
+            return None
         choices = event.get("choices")
-        if not isinstance(choices, list):
-            return
-        for choice in choices:
-            if not isinstance(choice, dict) or not isinstance(choice.get("delta"), dict):
-                continue
-            text = choice["delta"].get("reasoning_content")
-            if isinstance(text, str) and text:
-                self.trace("model.reasoning", request_id, choiceIndex=choice.get("index", 0), text=text)
+        if isinstance(choices, list):
+            for choice in choices:
+                if not isinstance(choice, dict) or not isinstance(choice.get("delta"), dict):
+                    continue
+                text = choice["delta"].get("reasoning_content")
+                if isinstance(text, str) and text:
+                    self.trace("model.reasoning", request_id, choiceIndex=choice.get("index", 0), text=text)
+        return normalize_usage(event.get("usage"))
 
     def connect(
         self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
@@ -119,7 +251,10 @@ async def request(
                 payload["model"] = self.config["modelId"]
                 url = self.config["url"].rstrip("/") + "/chat/completions"
                 request_id = secrets.token_hex(8)
+                self.request_count += 1
                 self.trace("model.request", request_id, payload=payload)
+                usage: dict[str, int] | None = None
+                response_status = 0
                 async with httpx.AsyncClient(
                     transport=self.transport,
                     event_hooks=MODEL_HTTP_HOOKS,
@@ -132,6 +267,7 @@ async def request(
                         headers=auth_headers(self.config),
                         json=payload,
                     ) as response:
+                        response_status = response.status_code
                         self.trace(
                             "model.response", request_id,
                             status=response.status_code,
@@ -145,6 +281,13 @@ async def request(
                         )
                         status_sent = True
                         await writer.drain()
+                        is_sse = "text/event-stream" in content_type
+                        collect_json = (
+                            not is_sse
+                            and 200 <= response.status_code < 300
+                            and "json" in content_type.lower()
+                        )
+                        json_body = bytearray() if collect_json else None
                         decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
                         buffer = ""
                         event_lines = []
@@ -155,7 +298,13 @@ async def request(
                             writer.write(f"{len(chunk):x}\r\n".encode() + chunk + b"\r\n")
                             await writer.drain()
                             forwarded_bytes += len(chunk)
-                            if "text/event-stream" not in content_type:
+                            if json_body is not None:
+                                if len(json_body) + len(chunk) <= MAX_JSON_USAGE_BYTES:
+                                    json_body.extend(chunk)
+                                else:
+                                    json_body = None
+                                continue
+                            if not is_sse:
                                 continue
                             buffer += decoder.decode(chunk)
                             while "\n" in buffer:
@@ -167,13 +316,33 @@ async def request(
                                     data = "\n".join(event_lines)
                                     event_lines.clear()
                                     if data != "[DONE]":
-                                        self.trace_reasoning(request_id, data)
+                                        observed = self.trace_reasoning(request_id, data)
+                                        usage = _merge_usage(usage, observed)
+                        if is_sse:
+                            buffer += decoder.decode(b"", final=True)
+                            for line in buffer.splitlines():
+                                line = line.rstrip("\r")
+                                if line.startswith("data:"):
+                                    event_lines.append(line[5:].lstrip(" "))
+                            if event_lines:
+                                data = "\n".join(event_lines)
+                                event_lines.clear()
+                                if data != "[DONE]":
+                                    observed = self.trace_reasoning(request_id, data)
+                                    usage = _merge_usage(usage, observed)
+                        elif json_body:
+                            with suppress(UnicodeDecodeError, json.JSONDecodeError):
+                                parsed = json.loads(json_body.decode("utf-8"))
+                                if isinstance(parsed, dict):
+                                    usage = normalize_usage(parsed.get("usage"))
                 # Only successful exhaustion and cleanup may terminate this response.
                 # Sending this in finally would disguise a truncated body as complete.
                 writer.write(b"0\r\n\r\n")
                 await writer.drain()
+                self._record_completion(response_status, usage)
                 self.trace("model.completed", request_id, forwardedBytes=forwarded_bytes,
-                           elapsedMs=round((monotonic() - started) * 1000))
+                           elapsedMs=round((monotonic() - started) * 1000),
+                           **({"usage": usage} if usage is not None else {}))
         except asyncio.CancelledError:
             if request_id is not None:
                 self.trace("model.cancelled", request_id, forwardedBytes=forwarded_bytes)

packages/core/src/opendirector/agent/native.py

@@ -357,6 +357,7 @@ async def observe(raw):
                         for task in model.active:
                             task.cancel()
                         await asyncio.gather(*model.active, return_exceptions=True)
+                        model.close()
                     if runtime == 'deepseek' and pump and not pump.done():
                         while not queue.empty():
                             queue.get_nowait()