Add pipeline trace ledger

.env.example

@@ -24,6 +24,14 @@ CK_DB_ADMIN_PG_SYSTEM_USER=postgres
 CK_DB_ADMIN_PSQL="runuser -u postgres -- psql -d creation_knowledge_prod -v ON_ERROR_STOP=1"
 CK_DB_MIGRATION_APPLY="ssh -i <key> -o BatchMode=yes -o ConnectTimeout=8 -o StrictHostKeyChecking=accept-new root@<host> \"runuser -u postgres -- psql -d creation_knowledge_prod -v ON_ERROR_STOP=1\" < path/to/migration.sql"
 
+# -----------------------------------------------------------------------------
+# Pipeline tracing ledger
+# -----------------------------------------------------------------------------
+CK_TRACE_ENABLED=true
+CK_TRACE_CAPTURE_REQUEST=true
+CK_TRACE_CAPTURE_RESPONSE=true
+CK_TRACE_MAX_PAYLOAD_CHARS=50000
+
 # -----------------------------------------------------------------------------
 # Upstream Open AIGC / category-tree database
 # Used for scope tree dumping only. Do not use this as the formal state DB.

acquisition/classification/coarse.py

@@ -16,6 +16,7 @@ from acquisition.classify import (
 from core.config import Settings
 from core.prompts import load_prompt
 from core.text_limits import CLASSIFY_BODY_MAX_CHARS, clip_text
+from pipeline.tracing import TraceContext, TraceWriter
 
 ROOT = Path(__file__).resolve().parents[2]
 
@@ -43,7 +44,13 @@ def prompt_version(*names: str) -> str:
     return h.hexdigest()[:16]
 
 
-def classify_imgtext(payload: dict[str, Any], settings: Settings) -> tuple:
+def classify_imgtext(
+    payload: dict[str, Any],
+    settings: Settings,
+    *,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
+) -> tuple:
     """Classify image-text content, accepting both HTTP image URLs and /data paths."""
     user = [
         {
@@ -67,11 +74,35 @@ def classify_imgtext(payload: dict[str, Any], settings: Settings) -> tuple:
         {"role": "system", "content": load_prompt("classify_imgtext")},
         {"role": "user", "content": user},
     ]
-    return _judge(messages, settings, timeout=120)
+    if trace_writer is None and trace_context is None:
+        return _judge(messages, settings, timeout=120)
+    return _judge(
+        messages,
+        settings,
+        timeout=120,
+        trace_writer=trace_writer,
+        trace_context=trace_context,
+        trace_stage="classify",
+        trace_substage="coarse_imgtext",
+        prompt_name="classify_imgtext",
+    )
 
 
-def classify_video(payload: dict[str, Any], settings: Settings) -> tuple:
-    return _legacy_classify_video(payload, settings)
+def classify_video(
+    payload: dict[str, Any],
+    settings: Settings,
+    *,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
+) -> tuple:
+    if trace_writer is None and trace_context is None:
+        return _legacy_classify_video(payload, settings)
+    return _legacy_classify_video(
+        payload,
+        settings,
+        trace_writer=trace_writer,
+        trace_context=trace_context,
+    )
 
 
 def coarse_classify_item(
@@ -83,6 +114,8 @@ def coarse_classify_item(
     image_urls: list[str] | None = None,
     video_url: str = "",
     settings: Settings,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> ClassificationResult:
     if content_mode == "unsupported":
         return ClassificationResult(
@@ -108,26 +141,38 @@ def coarse_classify_item(
         )
     if content_mode == "video_post" or (content_mode is None and video_url):
         version = prompt_version("classify_video")
-        is_creation, reason, knowledge, points = classify_video(
-            {
-                "platform": platform,
-                "title": title,
-                "body_text": body_text,
-                "video": video_url,
-            },
-            settings,
-        )
+        payload = {
+            "platform": platform,
+            "title": title,
+            "body_text": body_text,
+            "video": video_url,
+        }
+        if trace_writer is None and trace_context is None:
+            is_creation, reason, knowledge, points = classify_video(payload, settings)
+        else:
+            is_creation, reason, knowledge, points = classify_video(
+                payload,
+                settings,
+                trace_writer=trace_writer,
+                trace_context=trace_context,
+            )
     else:
         version = prompt_version("classify_imgtext")
-        is_creation, reason, knowledge, points = classify_imgtext(
-            {
-                "platform": platform,
-                "title": title,
-                "body_text": body_text,
-                "images": image_urls or [],
-            },
-            settings,
-        )
+        payload = {
+            "platform": platform,
+            "title": title,
+            "body_text": body_text,
+            "images": image_urls or [],
+        }
+        if trace_writer is None and trace_context is None:
+            is_creation, reason, knowledge, points = classify_imgtext(payload, settings)
+        else:
+            is_creation, reason, knowledge, points = classify_imgtext(
+                payload,
+                settings,
+                trace_writer=trace_writer,
+                trace_context=trace_context,
+            )
 
     if is_creation is None:
         return ClassificationResult(

acquisition/classify.py

@@ -31,6 +31,7 @@ from core.text_limits import (
     REASON_MAX_CHARS,
     clip_text,
 )
+from pipeline.tracing import TraceContext, TraceWriter, hash_prompt, redact_headers, timed_ms
 
 ROOT = Path(__file__).resolve().parent.parent
 PLATFORMS = ["xiaohongshu", "weixin", "douyin"]   # 默认重判全部；可传平台名覆盖
@@ -269,7 +270,17 @@ def _providers(settings: Settings, messages: list) -> list[tuple[str, str, dict,
     return out
 
 
-def _judge(messages: list, settings: Settings, timeout: float) -> tuple:
+def _judge(
+    messages: list,
+    settings: Settings,
+    timeout: float,
+    *,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
+    trace_stage: str = "classify",
+    trace_substage: str = "coarse",
+    prompt_name: str = "classify",
+) -> tuple:
     """调多模态模型（Qwen / OpenRouter / Ark），解析 {is_empty, reason, knowledge}，带重试。
     返回 (is_creation 1/0/None, reason, knowledge, points)。"""
     last = ""
@@ -279,18 +290,77 @@ def _judge(messages: list, settings: Settings, timeout: float) -> tuple:
         throttle = _provider_throttle(provider)
         min_interval = _provider_min_interval(provider, env)
         for attempt in range(3):
+            started = time.perf_counter()
             try:
                 throttle.wait(min_interval)
                 resp = httpx.post(api, headers=headers, json=payload, timeout=timeout)
                 if resp.status_code == 200:
-                    return _parse_judge_content(resp.json()["choices"][0]["message"]["content"])
+                    response_json = resp.json()
+                    content = response_json["choices"][0]["message"]["content"]
+                    parsed = _parse_judge_content(content)
+                    if trace_writer is not None:
+                        trace_writer.llm_call(
+                            context=trace_context or TraceContext(stage=trace_stage, substage=trace_substage),
+                            stage=trace_stage,
+                            substage=trace_substage,
+                            provider=name,
+                            model_name=payload.get("model"),
+                            endpoint=api,
+                            prompt_name=prompt_name,
+                            prompt_hash=hash_prompt(str(messages[0].get("content") if messages else "")),
+                            request_payload={"headers": redact_headers(headers), **payload},
+                            response_payload=response_json,
+                            parsed_payload={
+                                "is_creation": parsed[0],
+                                "reason": parsed[1],
+                                "knowledge": parsed[2],
+                                "points": parsed[3],
+                            },
+                            status="done",
+                            latency_ms=timed_ms(started),
+                            attempt_index=attempt + 1,
+                        )
+                    return parsed
                 last = f"{name} http {resp.status_code}"
+                if trace_writer is not None:
+                    trace_writer.llm_call(
+                        context=trace_context or TraceContext(stage=trace_stage, substage=trace_substage),
+                        stage=trace_stage,
+                        substage=trace_substage,
+                        provider=name,
+                        model_name=payload.get("model"),
+                        endpoint=api,
+                        prompt_name=prompt_name,
+                        prompt_hash=hash_prompt(str(messages[0].get("content") if messages else "")),
+                        request_payload={"headers": redact_headers(headers), **payload},
+                        response_payload={"http_status": resp.status_code, "text": resp.text},
+                        status="failed",
+                        error_message=last,
+                        latency_ms=timed_ms(started),
+                        attempt_index=attempt + 1,
+                    )
                 if resp.status_code == 429:
                     throttle.penalize(_provider_429_backoff(provider, env, attempt, resp))
                 if resp.status_code in (401, 403, 404):
                     break
             except Exception as exc:
                 last = f"{name} {clip_text(exc, ERROR_MESSAGE_MAX_CHARS)}"
+                if trace_writer is not None:
+                    trace_writer.llm_call(
+                        context=trace_context or TraceContext(stage=trace_stage, substage=trace_substage),
+                        stage=trace_stage,
+                        substage=trace_substage,
+                        provider=name,
+                        model_name=payload.get("model"),
+                        endpoint=api,
+                        prompt_name=prompt_name,
+                        prompt_hash=hash_prompt(str(messages[0].get("content") if messages else "")),
+                        request_payload={"headers": redact_headers(headers), **payload},
+                        status="failed",
+                        error_message=last,
+                        latency_ms=timed_ms(started),
+                        attempt_index=attempt + 1,
+                    )
             if "http " not in last or any(f"http {code}" in last for code in RETRYABLE_STATUS_CODES):
                 time.sleep(2 * (attempt + 1))
             else:
@@ -300,7 +370,13 @@ def _judge(messages: list, settings: Settings, timeout: float) -> tuple:
     return None, f"判定失败: {last}", "", ""
 
 
-def classify_imgtext(p: dict, settings: Settings) -> tuple:
+def classify_imgtext(
+    p: dict,
+    settings: Settings,
+    *,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
+) -> tuple:
     """图文：标题+正文+全部图，用收紧的 classify_imgtext.txt 判 is_empty 并提取知识点。"""
     user = [{"type": "text", "text": f"平台：{p.get('platform')}\n标题：{p.get('title', '')}\n"
              f"正文：{clip_text(p.get('body_text') or '', CLASSIFY_BODY_MAX_CHARS)}\n（下附帖子图片，请一并看完）"}]
@@ -313,10 +389,27 @@ def classify_imgtext(p: dict, settings: Settings) -> tuple:
             user.append({"type": "image_url", "image_url": {"url": u}})
     messages = [{"role": "system", "content": load_prompt("classify_imgtext")},
                 {"role": "user", "content": user}]
-    return _judge(messages, settings, timeout=120)
+    if trace_writer is None and trace_context is None:
+        return _judge(messages, settings, timeout=120)
+    return _judge(
+        messages,
+        settings,
+        timeout=120,
+        trace_writer=trace_writer,
+        trace_context=trace_context,
+        trace_stage="classify",
+        trace_substage="coarse_imgtext",
+        prompt_name="classify_imgtext",
+    )
 
 
-def classify_video(p: dict, settings: Settings) -> tuple:
+def classify_video(
+    p: dict,
+    settings: Settings,
+    *,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
+) -> tuple:
     """视频：看完整段视频，用 classify_video.txt 判 is_empty 并提取知识点。
 
     新采集链路里抖音视频会先转存 OSS，传入 HTTP(S) CDN URL；老链路传 /data 本地 mp4。
@@ -346,7 +439,18 @@ def classify_video(p: dict, settings: Settings) -> tuple:
     messages = [{"role": "system", "content": load_prompt("classify_video")},
                 {"role": "user", "content": [{"type": "text", "text": user_text},
                                              {"type": "video_url", "video_url": {"url": media}}]}]
-    return _judge(messages, settings, timeout=300)
+    if trace_writer is None and trace_context is None:
+        return _judge(messages, settings, timeout=300)
+    return _judge(
+        messages,
+        settings,
+        timeout=300,
+        trace_writer=trace_writer,
+        trace_context=trace_context,
+        trace_stage="classify",
+        trace_substage="coarse_video",
+        prompt_name="classify_video",
+    )
 
 
 if __name__ == "__main__":

acquisition/runner.py

@@ -21,6 +21,7 @@ from core.text_limits import (
     RAW_SUMMARY_MAX_CHARS,
     clip_text,
 )
+from pipeline.tracing import NoopTraceWriter, TraceContext, TraceWriter
 
 DEFAULT_PLATFORMS = ("xiaohongshu", "weixin", "douyin")
 PAGINATION_STRATEGY = "creation_ratio_two_page_v1"
@@ -182,6 +183,8 @@ def _record_candidate(
     classifier: Classifier,
     classify: bool,
     attempt_index: int = 1,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> RecordCandidateResult:
     source_payload = _source_payload(candidate, detail)
     platform_item_id = detail.source_id or candidate.source_id or None
@@ -232,8 +235,32 @@ def _record_candidate(
     )
     if item.id is None:
         raise RuntimeError("repository returned candidate without id")
+    _record_candidate_hit(
+        trace_writer,
+        trace_context,
+        item_id=item.id,
+        platform=platform,
+        unique_key=item.unique_key,
+        platform_item_id=item.platform_item_id,
+        search_provider=search_provider,
+        detail_provider=detail_provider,
+        candidate=candidate,
+        attempt_index=attempt_index,
+        is_duplicate_hit=False,
+        metadata={"content_mode": content_mode, "status": item.status},
+    )
 
     if not guard.can_process:
+        _trace_event(
+            trace_writer,
+            trace_context.child(item_id=item.id, platform=platform) if trace_context else None,
+            stage="classify",
+            event_type="item_skipped_before_classification",
+            status="skipped",
+            target_table="candidate_items",
+            target_id=item.id,
+            payload={"reason": guard.reason, "content_mode": content_mode},
+        )
         repo.add_item_classification(
             item_id=item.id,
             is_creation_knowledge=None,
@@ -296,18 +323,36 @@ def _record_candidate(
             status="skipped",
             error_message="video_oss_failed",
         )
+        _trace_event(
+            trace_writer,
+            trace_context.child(item_id=item.id, platform=platform) if trace_context else None,
+            stage="classify",
+            event_type="item_skipped_before_classification",
+            status="skipped",
+            target_table="candidate_items",
+            target_id=item.id,
+            payload={"reason": "video_oss_failed", "content_mode": content_mode},
+        )
         return RecordCandidateResult(displayed=True, skipped_processing=True)
 
     if classify:
-        result = classifier(
-            platform=platform,
-            content_mode=content_mode,
-            title=detail.title,
-            body_text=detail.body_text,
-            image_urls=_image_urls(media_rows),
-            video_url=_best_video_url(media_rows),
-            settings=settings,
-        )
+        classifier_kwargs = {
+            "platform": platform,
+            "content_mode": content_mode,
+            "title": detail.title,
+            "body_text": detail.body_text,
+            "image_urls": _image_urls(media_rows),
+            "video_url": _best_video_url(media_rows),
+            "settings": settings,
+        }
+        if classifier is coarse_classify_item:
+            classifier_kwargs.update(
+                {
+                    "trace_writer": trace_writer,
+                    "trace_context": trace_context.child(item_id=item.id, platform=platform) if trace_context else None,
+                }
+            )
+        result = classifier(**classifier_kwargs)
         repo.add_item_classification(
             item_id=item.id,
             is_creation_knowledge=result.is_creation_knowledge,
@@ -320,6 +365,21 @@ def _record_candidate(
             status=result.status,
             error_message=result.error_message,
         )
+        _trace_event(
+            trace_writer,
+            trace_context.child(item_id=item.id, platform=platform) if trace_context else None,
+            stage="classify",
+            event_type="classified",
+            status=result.status,
+            target_table="candidate_items",
+            target_id=item.id,
+            payload={
+                "is_creation_knowledge": result.is_creation_knowledge,
+                "label": result.label,
+                "confidence": result.confidence,
+            },
+            error_message=result.error_message,
+        )
         return RecordCandidateResult(
             displayed=True,
             classified=result.status == "classified" and result.is_creation_knowledge is not None,
@@ -356,6 +416,8 @@ def _attach_existing_candidate(
     candidate: PlatformCandidate,
     unique_key: str,
     attempt_index: int = 1,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> bool:
     get_by_key = getattr(repo, "get_candidate_item_by_unique_key", None)
     attach = getattr(repo, "attach_existing_candidate_item", None)
@@ -378,9 +440,83 @@ def _attach_existing_candidate(
             **({"search_provider": _candidate_provider(candidate)} if _candidate_provider(candidate) else {}),
         },
     )
+    _record_candidate_hit(
+        trace_writer,
+        trace_context,
+        item_id=existing.id,
+        platform=candidate.platform,
+        unique_key=unique_key,
+        platform_item_id=candidate.source_id or None,
+        search_provider=_candidate_provider(candidate),
+        detail_provider=None,
+        candidate=candidate,
+        attempt_index=attempt_index,
+        is_duplicate_hit=True,
+        metadata={"acquisition_match_status": "existing"},
+    )
+    _trace_event(
+        trace_writer,
+        trace_context.child(item_id=existing.id, platform=candidate.platform) if trace_context else None,
+        stage="search",
+        event_type="duplicate_candidate_reused",
+        status="skipped",
+        target_table="candidate_items",
+        target_id=existing.id,
+        payload={"unique_key": unique_key},
+    )
     return True
 
 
+def _trace_event(
+    trace_writer: TraceWriter | None,
+    context: TraceContext | None,
+    *,
+    stage: str,
+    event_type: str,
+    **kwargs: Any,
+) -> None:
+    if trace_writer is None or context is None:
+        return
+    trace_writer.event(context=context, stage=stage, event_type=event_type, **kwargs)
+
+
+def _record_candidate_hit(
+    trace_writer: TraceWriter | None,
+    context: TraceContext | None,
+    *,
+    item_id: UUID | None,
+    platform: str,
+    unique_key: str | None,
+    platform_item_id: str | None,
+    search_provider: str | None,
+    detail_provider: str | None,
+    candidate: PlatformCandidate,
+    attempt_index: int,
+    is_duplicate_hit: bool,
+    metadata: dict[str, Any] | None = None,
+) -> None:
+    if trace_writer is None or context is None:
+        return
+    raw = candidate.model_dump() if hasattr(candidate, "model_dump") else {}
+    trace_writer.candidate_hit(
+        context=context.child(item_id=item_id, platform=platform),
+        item_id=item_id,
+        platform=platform,
+        unique_key=unique_key,
+        platform_item_id=platform_item_id,
+        search_provider=search_provider,
+        detail_provider=detail_provider,
+        attempt_index=attempt_index,
+        page_index=raw.get("raw", {}).get("page_index") if isinstance(raw.get("raw"), dict) else None,
+        page_rank=raw.get("raw", {}).get("page_rank") if isinstance(raw.get("raw"), dict) else None,
+        candidate_rank=candidate.rank,
+        source_cursor=raw.get("raw", {}).get("source_cursor") if isinstance(raw.get("raw"), dict) else None,
+        is_duplicate_hit=is_duplicate_hit,
+        raw_candidate=raw,
+        metadata=metadata or {},
+    )
+
+
 def _pagination_config(threshold: float, max_pages: int) -> dict[str, Any]:
     return {
         "strategy": PAGINATION_STRATEGY,
@@ -449,6 +585,8 @@ def run_batch(
     pagination_max_pages: int = PAGINATION_MAX_PAGES,
     low_result_retry_threshold: int = LOW_RESULT_RETRY_THRESHOLD,
     low_result_max_retries: int = LOW_RESULT_MAX_RETRIES,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> RunBatchResult:
     """Run query x platform acquisition and write formal cloud-state rows."""
     queries = repo.list_queries_for_batch(batch_id, keep=True)
@@ -474,6 +612,21 @@ def run_batch(
     )
     if run.id is None:
         raise RuntimeError("repository returned acquisition run without id")
+    trace_writer = trace_writer or NoopTraceWriter()
+    run_trace_context = (trace_context or TraceContext()).child(
+        acquisition_run_id=run.id,
+        stage="search",
+    )
+    _trace_event(
+        trace_writer,
+        run_trace_context,
+        stage="search",
+        event_type="acquisition_started",
+        status="running",
+        target_table="acquisition_runs",
+        target_id=run.id,
+        payload={"batch_id": str(batch_id), "platforms": list(platforms)},
+    )
 
     total_jobs = len(queries) * len(platforms)
     done = partial = failed = skipped = 0
@@ -492,6 +645,19 @@ def run_batch(
                 metadata={"query_text": query.query_text},
             )
             if skip_done and job.status == "done":
+                _trace_event(
+                    trace_writer,
+                    run_trace_context.child(
+                        acquisition_job_id=_job_id(job),
+                        query_id=query_id,
+                        platform=platform,
+                    ),
+                    stage="search",
+                    event_type="job_skipped_done",
+                    status="skipped",
+                    target_table="acquisition_jobs",
+                    target_id=_job_id(job),
+                )
                 skipped += 1
                 continue
 
@@ -502,6 +668,21 @@ def run_batch(
                 attempt_count=attempts,
                 error_message=None,
             )
+            job_trace_context = run_trace_context.child(
+                acquisition_job_id=_job_id(job),
+                query_id=query_id,
+                platform=platform,
+            )
+            _trace_event(
+                trace_writer,
+                job_trace_context,
+                stage="search",
+                event_type="job_started",
+                status="running",
+                target_table="acquisition_jobs",
+                target_id=_job_id(job),
+                payload={"query_text": query.query_text, "platform": platform},
+            )
             errors: list[str] = []
             display_count = 0
             searched_count = 0
@@ -578,6 +759,8 @@ def run_batch(
                                     candidate=candidate,
                                     unique_key=unique_key,
                                     attempt_index=attempt_index,
+                                    trace_writer=trace_writer,
+                                    trace_context=job_trace_context,
                                 ):
                                     display_count += 1
                                     continue
@@ -598,6 +781,8 @@ def run_batch(
                                     classifier=classifier,
                                     classify=classify,
                                     attempt_index=attempt_index,
+                                    trace_writer=trace_writer,
+                                    trace_context=job_trace_context,
                                 )
                                 if recorded.displayed:
                                     display_count += 1
@@ -608,7 +793,21 @@ def run_batch(
                                     if recorded.is_creation_knowledge is True:
                                         page_creation_count += 1
                             except Exception as exc:
-                                errors.append(clip_text(exc, ERROR_MESSAGE_MAX_CHARS))
+                                message = clip_text(exc, ERROR_MESSAGE_MAX_CHARS)
+                                errors.append(message)
+                                _trace_event(
+                                    trace_writer,
+                                    job_trace_context,
+                                    stage="search",
+                                    event_type="candidate_failed",
+                                    status="failed",
+                                    severity="warning",
+                                    payload={
+                                        "candidate": candidate.model_dump() if hasattr(candidate, "model_dump") else {},
+                                        "attempt_index": attempt_index,
+                                    },
+                                    error_message=message,
+                                )
                                 continue
 
                         page_ratio = _ratio(page_creation_count, page_classified_count)
@@ -627,6 +826,15 @@ def run_batch(
                         }
                         page_summaries.append(page_summary)
                         attempt_page_summaries.append(page_summary)
+                        _trace_event(
+                            trace_writer,
+                            job_trace_context,
+                            stage="search",
+                            event_type="page_searched",
+                            status="done",
+                            payload=page_summary,
+                            attempt_index=attempt_index,
+                        )
 
                         if page.page_index == 1:
                             attempt_first_page_classified_count = page_classified_count
@@ -722,13 +930,31 @@ def run_batch(
                         "errors": errors[-3:],
                     },
                 )
+                _trace_event(
+                    trace_writer,
+                    job_trace_context,
+                    stage="search",
+                    event_type="job_finished",
+                    status=status,
+                    target_table="acquisition_jobs",
+                    target_id=_job_id(job),
+                    payload={
+                        "searched_count": searched_count,
+                        "raw_searched_count": raw_searched_count,
+                        "display_count": display_count,
+                        "skipped_item_count": skipped_item_count,
+                        "search_page_count": search_page_count,
+                    },
+                    error_message=None if status != "failed" else "; ".join(errors[-3:]),
+                )
             except Exception as exc:
+                message = clip_text(exc, ERROR_MESSAGE_MAX_CHARS)
                 failed += 1
                 repo.update_acquisition_job(
                     _job_id(job),
                     status="failed",
                     attempt_count=attempts,
-                    error_message=clip_text(exc, ERROR_MESSAGE_MAX_CHARS),
+                    error_message=message,
                     metadata={
                         "query_text": query.query_text,
                         "searched_count": searched_count,
@@ -751,6 +977,17 @@ def run_batch(
                         "errors": errors[-3:],
                     },
                 )
+                _trace_event(
+                    trace_writer,
+                    job_trace_context,
+                    stage="search",
+                    event_type="job_failed",
+                    status="failed",
+                    severity="error",
+                    target_table="acquisition_jobs",
+                    target_id=_job_id(job),
+                    error_message=message,
+                )
 
     run_status = "done" if done > 0 else "failed"
     update_run = getattr(repo, "update_acquisition_run", None)
@@ -770,6 +1007,22 @@ def run_batch(
                 ),
             },
         )
+    _trace_event(
+        trace_writer,
+        run_trace_context,
+        stage="search",
+        event_type="acquisition_finished",
+        status=run_status,
+        target_table="acquisition_runs",
+        target_id=run.id,
+        payload={
+            "total_jobs": total_jobs,
+            "done": done,
+            "partial": partial,
+            "failed": failed,
+            "skipped": skipped,
+        },
+    )
 
     return RunBatchResult(
         run_id=run.id,

app/dependencies.py

@@ -4,12 +4,13 @@ from __future__ import annotations
 import os
 from typing import Iterator
 
-from fastapi import Depends, HTTPException
+from fastapi import Depends
 
 from acquisition.repositories.postgres import PostgresAcquisitionRepository
 from core.config import CreationDbConfig
 from core.db_session import pooled_transaction
 from decode_content.repositories.postgres import PostgresDecodeRepository
+from pipeline.postgres import PostgresPipelineRepository
 
 
 def _env_file() -> str:
@@ -33,5 +34,5 @@ def get_decode_repository(conn: object = Depends(get_db_connection)) -> Postgres
     return PostgresDecodeRepository(conn)
 
 
-def get_pipeline_repository():
-    raise HTTPException(status_code=501, detail="pipeline repository persistence is wired in Step 7")
+def get_pipeline_repository(conn: object = Depends(get_db_connection)) -> PostgresPipelineRepository:
+    return PostgresPipelineRepository(conn)

app/routes/manual_queries.py

@@ -7,6 +7,7 @@ import sys
 from datetime import datetime
 from pathlib import Path
 from typing import Any
+from uuid import UUID
 
 from fastapi import APIRouter, Body, Depends, HTTPException
 from pydantic import BaseModel, Field
@@ -16,6 +17,8 @@ from app.dependencies import _env_file, get_creation_db_config
 from app.routes.acquisition import _model_dump
 from core.config import CreationDbConfig
 from core.db_session import transaction
+from pipeline.postgres import PostgresPipelineRepository
+from pipeline.tracing import TraceContext, new_trace_writer
 
 router = APIRouter(prefix="/api/query-batches", tags=["manual-query-batches"])
 
@@ -170,6 +173,38 @@ def _pipeline_command(
     return cmd
 
 
+def _create_pipeline_trace(
+    db_config: CreationDbConfig,
+    *,
+    batch_id: Any,
+    run_key: str,
+    platforms: list[str],
+    request: ManualQueryBatchRequest,
+    log_path: Path,
+) -> str | None:
+    try:
+        with transaction(db_config) as conn:
+            run = PostgresPipelineRepository(conn).create_pipeline_run(
+                run_key=run_key,
+                batch_id=batch_id,
+                status="pending",
+                current_stage="query",
+                config={
+                    "platforms": platforms,
+                    "search_limit": request.search_limit,
+                    "display_limit": request.display_limit,
+                    "decode_limit": request.decode_limit,
+                },
+                metadata={
+                    "source": "manual_api",
+                    "log_path": str(log_path),
+                },
+            )
+            return str(run.id) if run.id else None
+    except Exception:
+        return None
+
+
 @router.post("/manual")
 def create_manual_query_batch(
     payload: Any = Body(...),
@@ -241,6 +276,28 @@ def create_manual_query_batch(
             metadata=run_metadata,
         )
 
+    pipeline_run_id = _create_pipeline_trace(
+        db_config,
+        batch_id=batch.id,
+        run_key=run_key,
+        platforms=platforms,
+        request=request,
+        log_path=log_path,
+    )
+    if pipeline_run_id:
+        trace_writer = new_trace_writer(db_config, env_file=_env_file())
+        trace_writer.event(
+            context=TraceContext(
+                pipeline_run_id=UUID(pipeline_run_id),
+                acquisition_run_id=run.id,
+                stage="query",
+            ),
+            stage="query",
+            event_type="manual_query_queued",
+            status="pending",
+            payload={"query_count": len(queries), "platforms": platforms},
+        )
+
     RUNTIME_MANUAL_DIR.mkdir(parents=True, exist_ok=True)
     env = os.environ.copy()
     env["PYTHONPATH"] = f"{ROOT}:{env.get('PYTHONPATH', '')}".rstrip(":")
@@ -254,6 +311,19 @@ def create_manual_query_batch(
                 stderr=subprocess.STDOUT,
                 start_new_session=True,
             )
+        if pipeline_run_id:
+            trace_writer = new_trace_writer(db_config, env_file=_env_file())
+            trace_writer.event(
+                context=TraceContext(
+                    pipeline_run_id=UUID(pipeline_run_id),
+                    acquisition_run_id=run.id,
+                    stage="query",
+                ),
+                stage="query",
+                event_type="process_started",
+                status="running",
+                payload={"pid": process.pid, "log_path": str(log_path)},
+            )
     except OSError as exc:
         with transaction(db_config) as conn:
             PostgresAcquisitionRepository(conn).update_acquisition_run(
@@ -267,6 +337,7 @@ def create_manual_query_batch(
     return {
         "status": "queued",
         "batch_id": str(batch.id),
+        "pipeline_run_id": pipeline_run_id,
         "run_id": str(run.id),
         "run_key": run_key,
         "pid": process.pid,

app/routes/runs.py

@@ -19,5 +19,47 @@ def pipeline_run(
     getter = getattr(repo, "get_pipeline_run", None)
     if getter is None:
         raise HTTPException(status_code=501, detail="pipeline run reader is not implemented")
-    run = getter(run_id)
+    try:
+        run = getter(run_id)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=404, detail="pipeline run not found") from exc
+    return {"run": run.model_dump(mode="json") if hasattr(run, "model_dump") else run}
+
+
+@router.get("/runs/{run_id}/timeline")
+def pipeline_run_timeline(
+    run_id: UUID,
+    repo: Any = Depends(get_pipeline_repository),
+) -> dict[str, Any]:
+    getter = getattr(repo, "get_pipeline_run", None)
+    list_timeline = getattr(repo, "list_timeline", None)
+    if getter is None or list_timeline is None:
+        raise HTTPException(status_code=501, detail="pipeline timeline reader is not implemented")
+    try:
+        run = getter(run_id)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=404, detail="pipeline run not found") from exc
+    bundle_getter = getattr(repo, "get_timeline_bundle", None)
+    if bundle_getter is not None:
+        bundle = bundle_getter(run_id)
+    else:
+        bundle = {"events": list_timeline(run_id), "jobs": [], "candidate_hits": [], "llm_call_traces": []}
+    return {
+        "run": run.model_dump(mode="json") if hasattr(run, "model_dump") else run,
+        **bundle,
+    }
+
+
+@router.get("/runs/by-acquisition/{acquisition_run_id}")
+def pipeline_run_by_acquisition(
+    acquisition_run_id: UUID,
+    repo: Any = Depends(get_pipeline_repository),
+) -> dict[str, Any]:
+    getter = getattr(repo, "get_pipeline_run_by_acquisition_run", None)
+    if getter is None:
+        raise HTTPException(status_code=501, detail="pipeline acquisition lookup is not implemented")
+    try:
+        run = getter(acquisition_run_id)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=404, detail="pipeline run not found") from exc
     return {"run": run.model_dump(mode="json") if hasattr(run, "model_dump") else run}

core/llm.py

@@ -4,12 +4,14 @@
 """
 from __future__ import annotations
 
+import time
 from typing import Any, Callable, Optional
 
 import httpx
 
 from core.config import Settings
 from core.jsonio import extract_json_object
+from pipeline.tracing import TraceContext, TraceWriter, hash_prompt, redact_headers, timed_ms
 
 
 class LLMError(RuntimeError):
@@ -25,6 +27,11 @@ def chat_json(
     http_post: Callable[..., Any] = httpx.post,
     env_file: str = ".env",
     timeout: float = 60.0,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
+    trace_stage: str = "decode",
+    trace_substage: str = "chat_json",
+    prompt_name: str = "chat_json",
 ) -> dict:
     """调一次对话，强约束输出 JSON，返回解析后的 dict。带一次重试。"""
     settings = settings or Settings.from_env(env_file)
@@ -44,25 +51,97 @@ def chat_json(
     last_exc: Optional[Exception] = None
     # 最多 3 轮：HTTP 错误重试；JSON 不合法则把坏输出喂回去做 self-repair
     for attempt in range(3):
+        started = time.perf_counter()
+        response_json: dict[str, Any] = {}
+        request_payload = {
+            "model": model,
+            "messages": messages,
+            "response_format": {"type": "json_object"},
+        }
         try:
-            resp = http_post(url, headers=headers, json={
-                "model": model,
-                "messages": messages,
-                "response_format": {"type": "json_object"},
-            }, timeout=timeout)
+            resp = http_post(url, headers=headers, json=request_payload, timeout=timeout)
             resp.raise_for_status()
-            content = resp.json()["choices"][0]["message"]["content"]
+            response_json = resp.json()
+            content = response_json["choices"][0]["message"]["content"]
         except httpx.HTTPError as exc:
             last_exc = exc
+            if trace_writer is not None:
+                trace_writer.llm_call(
+                    context=trace_context or TraceContext(stage=trace_stage, substage=trace_substage),
+                    stage=trace_stage,
+                    substage=trace_substage,
+                    provider="bailian",
+                    model_name=model,
+                    endpoint=url,
+                    prompt_name=prompt_name,
+                    prompt_hash=hash_prompt(system),
+                    request_payload={"headers": redact_headers(headers), **request_payload},
+                    status="failed",
+                    error_message=str(exc),
+                    latency_ms=timed_ms(started),
+                    attempt_index=attempt + 1,
+                )
             if attempt < 2:
                 continue
             raise LLMError(f"llm_http_error: {exc}") from exc
         except (KeyError, IndexError, TypeError) as exc:
+            if trace_writer is not None:
+                trace_writer.llm_call(
+                    context=trace_context or TraceContext(stage=trace_stage, substage=trace_substage),
+                    stage=trace_stage,
+                    substage=trace_substage,
+                    provider="bailian",
+                    model_name=model,
+                    endpoint=url,
+                    prompt_name=prompt_name,
+                    prompt_hash=hash_prompt(system),
+                    request_payload={"headers": redact_headers(headers), **request_payload},
+                    response_payload=response_json,
+                    status="failed",
+                    error_message=str(exc),
+                    latency_ms=timed_ms(started),
+                    attempt_index=attempt + 1,
+                )
             raise LLMError(f"llm_response_invalid: {exc}") from exc
         try:
-            return extract_json_object(content)
+            parsed = extract_json_object(content)
+            if trace_writer is not None:
+                trace_writer.llm_call(
+                    context=trace_context or TraceContext(stage=trace_stage, substage=trace_substage),
+                    stage=trace_stage,
+                    substage=trace_substage,
+                    provider="bailian",
+                    model_name=model,
+                    endpoint=url,
+                    prompt_name=prompt_name,
+                    prompt_hash=hash_prompt(system),
+                    request_payload={"headers": redact_headers(headers), **request_payload},
+                    response_payload=response_json,
+                    parsed_payload=parsed,
+                    status="done",
+                    latency_ms=timed_ms(started),
+                    attempt_index=attempt + 1,
+                )
+            return parsed
         except ValueError as exc:
             last_exc = exc
+            if trace_writer is not None:
+                trace_writer.llm_call(
+                    context=trace_context or TraceContext(stage=trace_stage, substage=trace_substage),
+                    stage=trace_stage,
+                    substage=trace_substage,
+                    provider="bailian",
+                    model_name=model,
+                    endpoint=url,
+                    prompt_name=prompt_name,
+                    prompt_hash=hash_prompt(system),
+                    request_payload={"headers": redact_headers(headers), **request_payload},
+                    response_payload={"content": content},
+                    status="failed",
+                    error_message=str(exc),
+                    latency_ms=timed_ms(started),
+                    attempt_index=attempt + 1,
+                )
             messages = messages + [
                 {"role": "assistant", "content": content},
                 {"role": "user", "content": (

db/migrations/001_creation_knowledge_schema.sql

@@ -220,15 +220,128 @@ CREATE TABLE IF NOT EXISTS creation_knowledge.ingest_records (
     updated_at timestamptz NOT NULL DEFAULT now()
 );
 
-CREATE TABLE IF NOT EXISTS creation_knowledge.contract_snapshots (
+CREATE TABLE IF NOT EXISTS creation_knowledge.pipeline_runs (
+    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    run_key text UNIQUE,
+    batch_id uuid REFERENCES creation_knowledge.query_batches(id) ON DELETE SET NULL,
+    status text NOT NULL DEFAULT 'pending',
+    current_stage text,
+    config jsonb NOT NULL DEFAULT '{}'::jsonb,
+    metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
+    error_message text,
+    started_at timestamptz,
+    finished_at timestamptz,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    updated_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS creation_knowledge.pipeline_jobs (
+    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    pipeline_run_id uuid NOT NULL REFERENCES creation_knowledge.pipeline_runs(id) ON DELETE CASCADE,
+    stage text NOT NULL,
+    status text NOT NULL DEFAULT 'pending',
+    target_table text,
+    target_id uuid,
+    attempt_count integer NOT NULL DEFAULT 0,
+    metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
+    error_message text,
+    started_at timestamptz,
+    finished_at timestamptz,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    updated_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS creation_knowledge.pipeline_run_events (
+    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    pipeline_run_id uuid REFERENCES creation_knowledge.pipeline_runs(id) ON DELETE CASCADE,
+    pipeline_job_id uuid REFERENCES creation_knowledge.pipeline_jobs(id) ON DELETE SET NULL,
+    stage text NOT NULL,
+    event_type text NOT NULL,
+    status text,
+    severity text NOT NULL DEFAULT 'info',
+    target_table text,
+    target_id uuid,
+    acquisition_run_id uuid REFERENCES creation_knowledge.acquisition_runs(id) ON DELETE SET NULL,
+    acquisition_job_id uuid REFERENCES creation_knowledge.acquisition_jobs(id) ON DELETE SET NULL,
+    query_id uuid REFERENCES creation_knowledge.queries(id) ON DELETE SET NULL,
+    item_id uuid REFERENCES creation_knowledge.candidate_items(id) ON DELETE SET NULL,
+    decode_job_id uuid REFERENCES creation_knowledge.decode_jobs(id) ON DELETE SET NULL,
+    decode_result_id uuid REFERENCES creation_knowledge.decode_results(id) ON DELETE SET NULL,
+    payload_draft_id uuid REFERENCES creation_knowledge.payload_drafts(id) ON DELETE SET NULL,
+    ingest_record_id uuid REFERENCES creation_knowledge.ingest_records(id) ON DELETE SET NULL,
+    platform text,
+    attempt_index integer,
+    message text,
+    payload jsonb NOT NULL DEFAULT '{}'::jsonb,
+    error_message text,
+    duration_ms integer,
+    created_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS creation_knowledge.candidate_item_hits (
+    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    pipeline_run_id uuid REFERENCES creation_knowledge.pipeline_runs(id) ON DELETE CASCADE,
+    acquisition_run_id uuid REFERENCES creation_knowledge.acquisition_runs(id) ON DELETE CASCADE,
+    acquisition_job_id uuid REFERENCES creation_knowledge.acquisition_jobs(id) ON DELETE SET NULL,
+    query_id uuid REFERENCES creation_knowledge.queries(id) ON DELETE SET NULL,
+    item_id uuid REFERENCES creation_knowledge.candidate_items(id) ON DELETE SET NULL,
+    platform text NOT NULL,
+    unique_key text,
+    platform_item_id text,
+    search_provider text,
+    detail_provider text,
+    attempt_index integer,
+    page_index integer,
+    page_rank integer,
+    candidate_rank integer,
+    source_cursor text,
+    is_duplicate_hit boolean NOT NULL DEFAULT false,
+    raw_candidate jsonb NOT NULL DEFAULT '{}'::jsonb,
+    metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
+    created_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS creation_knowledge.llm_call_traces (
+    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    pipeline_run_id uuid REFERENCES creation_knowledge.pipeline_runs(id) ON DELETE CASCADE,
+    pipeline_job_id uuid REFERENCES creation_knowledge.pipeline_jobs(id) ON DELETE SET NULL,
+    run_event_id uuid REFERENCES creation_knowledge.pipeline_run_events(id) ON DELETE SET NULL,
+    stage text NOT NULL,
+    substage text,
+    provider text,
+    model_name text,
+    endpoint text,
+    prompt_name text,
+    prompt_hash text,
+    trace_context jsonb NOT NULL DEFAULT '{}'::jsonb,
+    request_payload jsonb NOT NULL DEFAULT '{}'::jsonb,
+    response_payload jsonb NOT NULL DEFAULT '{}'::jsonb,
+    parsed_payload jsonb NOT NULL DEFAULT '{}'::jsonb,
+    status text NOT NULL DEFAULT 'pending',
+    error_message text,
+    latency_ms integer,
+    attempt_index integer,
+    created_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS creation_knowledge.contract_artifacts (
     id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
     contract_name text NOT NULL,
     contract_type text NOT NULL,
     version_label text,
-    content_hash text,
+    content_hash text NOT NULL,
     source_path text,
     snapshot jsonb NOT NULL DEFAULT '{}'::jsonb,
-    created_at timestamptz NOT NULL DEFAULT now()
+    created_at timestamptz NOT NULL DEFAULT now(),
+    updated_at timestamptz NOT NULL DEFAULT now(),
+    UNIQUE (contract_name, contract_type, source_path, content_hash)
+);
+
+CREATE TABLE IF NOT EXISTS creation_knowledge.run_contract_artifacts (
+    pipeline_run_id uuid NOT NULL REFERENCES creation_knowledge.pipeline_runs(id) ON DELETE CASCADE,
+    contract_artifact_id uuid NOT NULL REFERENCES creation_knowledge.contract_artifacts(id) ON DELETE CASCADE,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    PRIMARY KEY (pipeline_run_id, contract_artifact_id)
 );
 
 CREATE INDEX IF NOT EXISTS idx_queries_batch ON creation_knowledge.queries(batch_id);
@@ -244,7 +357,24 @@ CREATE INDEX IF NOT EXISTS idx_knowledge_particles_item ON creation_knowledge.kn
 CREATE INDEX IF NOT EXISTS idx_scope_results_particle ON creation_knowledge.scope_results(particle_id);
 CREATE INDEX IF NOT EXISTS idx_payload_drafts_particle ON creation_knowledge.payload_drafts(particle_id);
 CREATE INDEX IF NOT EXISTS idx_ingest_records_payload ON creation_knowledge.ingest_records(payload_draft_id);
-CREATE INDEX IF NOT EXISTS idx_contract_snapshots_name ON creation_knowledge.contract_snapshots(contract_name, contract_type);
+CREATE INDEX IF NOT EXISTS idx_pipeline_runs_batch ON creation_knowledge.pipeline_runs(batch_id);
+CREATE INDEX IF NOT EXISTS idx_pipeline_runs_status ON creation_knowledge.pipeline_runs(status, current_stage);
+CREATE INDEX IF NOT EXISTS idx_pipeline_jobs_run ON creation_knowledge.pipeline_jobs(pipeline_run_id);
+CREATE INDEX IF NOT EXISTS idx_pipeline_jobs_stage_status ON creation_knowledge.pipeline_jobs(stage, status);
+CREATE INDEX IF NOT EXISTS idx_pipeline_run_events_run ON creation_knowledge.pipeline_run_events(pipeline_run_id, created_at);
+CREATE INDEX IF NOT EXISTS idx_pipeline_run_events_acq_run ON creation_knowledge.pipeline_run_events(acquisition_run_id);
+CREATE INDEX IF NOT EXISTS idx_pipeline_run_events_query ON creation_knowledge.pipeline_run_events(query_id);
+CREATE INDEX IF NOT EXISTS idx_pipeline_run_events_item ON creation_knowledge.pipeline_run_events(item_id);
+CREATE INDEX IF NOT EXISTS idx_candidate_item_hits_pipeline_run ON creation_knowledge.candidate_item_hits(pipeline_run_id);
+CREATE INDEX IF NOT EXISTS idx_candidate_item_hits_acq_run ON creation_knowledge.candidate_item_hits(acquisition_run_id);
+CREATE INDEX IF NOT EXISTS idx_candidate_item_hits_query ON creation_knowledge.candidate_item_hits(query_id);
+CREATE INDEX IF NOT EXISTS idx_candidate_item_hits_item ON creation_knowledge.candidate_item_hits(item_id);
+CREATE INDEX IF NOT EXISTS idx_candidate_item_hits_unique_key ON creation_knowledge.candidate_item_hits(unique_key);
+CREATE INDEX IF NOT EXISTS idx_llm_call_traces_run_stage ON creation_knowledge.llm_call_traces(pipeline_run_id, stage, substage);
+CREATE INDEX IF NOT EXISTS idx_llm_call_traces_item ON creation_knowledge.llm_call_traces((trace_context ->> 'item_id'));
+CREATE INDEX IF NOT EXISTS idx_llm_call_traces_status ON creation_knowledge.llm_call_traces(status);
+CREATE INDEX IF NOT EXISTS idx_contract_artifacts_name_hash ON creation_knowledge.contract_artifacts(contract_name, contract_type, source_path, content_hash);
+CREATE INDEX IF NOT EXISTS idx_run_contract_artifacts_artifact ON creation_knowledge.run_contract_artifacts(contract_artifact_id);
 
 DO $$
 DECLARE
@@ -263,7 +393,10 @@ BEGIN
         'knowledge_particles',
         'scope_results',
         'payload_drafts',
-        'ingest_records'
+        'ingest_records',
+        'pipeline_runs',
+        'pipeline_jobs',
+        'contract_artifacts'
     ]
     LOOP
         EXECUTE format('DROP TRIGGER IF EXISTS trg_%I_touch_updated_at ON creation_knowledge.%I', table_name, table_name);

decode_content/contracts.py

@@ -275,9 +275,18 @@ def compute_contract_hash(contract: SkillContract | None = None) -> str:
     return (contract or load_contract()).content_hash
 
 
-def iter_contract_snapshots(
+def iter_contract_artifacts(
     contract: SkillContract | None = None,
     *,
     version_label: str | None = None,
 ) -> list[ContractSnapshot]:
     return (contract or load_contract()).snapshots(version_label=version_label)
+
+
+def iter_contract_snapshots(
+    contract: SkillContract | None = None,
+    *,
+    version_label: str | None = None,
+) -> list[ContractSnapshot]:
+    """Backward-compatible alias for tests and callers that need artifact payloads."""
+    return iter_contract_artifacts(contract, version_label=version_label)

decode_content/readers/imgtext.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import logging
+import time
 from typing import Any, Callable, Mapping, Optional
 
 import httpx
@@ -10,6 +11,7 @@ from core.config import load_env_file
 from core.jsonio import extract_json_object, to_bool
 from core.models import Card, CardExtract, ExtractedContent, Post
 from core.prompts import load_prompt
+from pipeline.tracing import TraceContext, TraceWriter, hash_prompt, redact_headers, timed_ms
 
 logger = logging.getLogger(__name__)
 
@@ -103,28 +105,59 @@ class BailianExtractor:
             {"role": "user", "content": parts},
         ]
 
-    def extract(self, post: Post) -> ExtractedContent:
+    def extract(
+        self,
+        post: Post,
+        *,
+        trace_writer: TraceWriter | None = None,
+        trace_context: TraceContext | None = None,
+    ) -> ExtractedContent:
         messages = self.build_messages(post)
         last_exc: Optional[Exception] = None
         for attempt in range(2):
+            started = time.perf_counter()
+            headers = {
+                "Authorization": f"Bearer {self.api_key}",
+                "Content-Type": "application/json",
+            }
+            body = {"model": self.model, "messages": messages}
             try:
                 resp = self.http_post(
                     f"{self.base_url}/chat/completions",
-                    headers={
-                        "Authorization": f"Bearer {self.api_key}",
-                        "Content-Type": "application/json",
-                    },
-                    json={"model": self.model, "messages": messages},
+                    headers=headers,
+                    json=body,
                     timeout=self.timeout_seconds,
                 )
                 resp.raise_for_status()
-                content = resp.json()["choices"][0]["message"]["content"]
+                response_json = resp.json()
+                content = response_json["choices"][0]["message"]["content"]
                 data = extract_json_object(content)
                 cards = [
                     CardExtract(index=int(card["index"]), content=str(card.get("content") or ""))
                     for card in (data.get("cards") or [])
                     if isinstance(card, dict) and card.get("index") is not None
                 ]
+                if trace_writer is not None:
+                    trace_writer.llm_call(
+                        context=trace_context or TraceContext(stage="decode", substage="read_imgtext"),
+                        stage="decode",
+                        substage="read_imgtext",
+                        provider="bailian",
+                        model_name=self.model,
+                        endpoint=f"{self.base_url}/chat/completions",
+                        prompt_name="extract_imgtext",
+                        prompt_hash=hash_prompt(SYSTEM_PROMPT),
+                        request_payload={"headers": redact_headers(headers), **body},
+                        response_payload=response_json,
+                        parsed_payload={
+                            "is_empty": to_bool(data.get("is_empty")),
+                            "card_count": len(cards),
+                            "text": data.get("text"),
+                        },
+                        status="done",
+                        latency_ms=timed_ms(started),
+                        attempt_index=attempt + 1,
+                    )
                 return ExtractedContent(
                     text=str(data.get("text") or ""),
                     cards=cards,
@@ -134,11 +167,43 @@ class BailianExtractor:
                 )
             except httpx.HTTPError as exc:
                 last_exc = exc
+                if trace_writer is not None:
+                    trace_writer.llm_call(
+                        context=trace_context or TraceContext(stage="decode", substage="read_imgtext"),
+                        stage="decode",
+                        substage="read_imgtext",
+                        provider="bailian",
+                        model_name=self.model,
+                        endpoint=f"{self.base_url}/chat/completions",
+                        prompt_name="extract_imgtext",
+                        prompt_hash=hash_prompt(SYSTEM_PROMPT),
+                        request_payload={"headers": redact_headers(headers), **body},
+                        status="failed",
+                        error_message=str(exc),
+                        latency_ms=timed_ms(started),
+                        attempt_index=attempt + 1,
+                    )
                 if attempt == 0:
                     continue
                 raise ExtractorError(f"bailian_http_error: {exc}") from exc
             except (KeyError, IndexError, TypeError, ValueError) as exc:
                 last_exc = exc
+                if trace_writer is not None:
+                    trace_writer.llm_call(
+                        context=trace_context or TraceContext(stage="decode", substage="read_imgtext"),
+                        stage="decode",
+                        substage="read_imgtext",
+                        provider="bailian",
+                        model_name=self.model,
+                        endpoint=f"{self.base_url}/chat/completions",
+                        prompt_name="extract_imgtext",
+                        prompt_hash=hash_prompt(SYSTEM_PROMPT),
+                        request_payload={"headers": redact_headers(headers), **body},
+                        status="failed",
+                        error_message=str(exc),
+                        latency_ms=timed_ms(started),
+                        attempt_index=attempt + 1,
+                    )
                 if attempt == 0:
                     continue
                 raise ExtractorError(f"bailian_response_invalid: {exc}") from exc
@@ -153,14 +218,22 @@ def extract_content(
     *,
     client: Optional[BailianExtractor] = None,
     env_file: str = ".env",
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> ExtractedContent:
     client = client or GeminiExtractor.from_env(env_file=env_file)
-    return client.extract(post)
+    return client.extract(post, trace_writer=trace_writer, trace_context=trace_context)
 
 
-def read_imgtext(post: Post, *, extractor: BailianExtractor | None = None) -> ExtractedContent:
+def read_imgtext(
+    post: Post,
+    *,
+    extractor: BailianExtractor | None = None,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
+) -> ExtractedContent:
     client = extractor or BailianExtractor.from_env()
-    return client.extract(post)
+    return client.extract(post, trace_writer=trace_writer, trace_context=trace_context)
 
 
 __all__ = [

decode_content/readers/service.py

@@ -10,6 +10,7 @@ from core.models import Card, ExtractedContent, Post
 from decode_content.models import ReadCard, ReadResult
 from decode_content.readers.imgtext import BailianExtractor, read_imgtext
 from decode_content.readers.video import read_video
+from pipeline.tracing import TraceContext, TraceWriter
 
 
 class ImageReader(Protocol):
@@ -176,6 +177,8 @@ def read_post(
     extractor: BailianExtractor | None = None,
     image_reader: ImageReader | None = None,
     video_reader: VideoReader | None = None,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> ReadResult:
     mode = post.content_mode or ("video_post" if post.video_urls else "image_post")
     if mode == "unsupported":
@@ -192,9 +195,16 @@ def read_post(
                 post,
                 settings=settings,
                 oss_video_url=post.video_urls[0],
+                trace_writer=trace_writer,
+                trace_context=trace_context,
             )
     else:
-        extracted = image_reader(post) if image_reader is not None else read_imgtext(post, extractor=extractor)
+        extracted = image_reader(post) if image_reader is not None else read_imgtext(
+            post,
+            extractor=extractor,
+            trace_writer=trace_writer,
+            trace_context=trace_context,
+        )
     return read_result_from_extracted(post, extracted)
 
 
@@ -206,6 +216,8 @@ def read_item(
     extractor: BailianExtractor | None = None,
     image_reader: ImageReader | None = None,
     video_reader: VideoReader | None = None,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> ReadResult:
     return read_post(
         post_from_candidate_item(item, media_assets),
@@ -213,6 +225,8 @@ def read_item(
         extractor=extractor,
         image_reader=image_reader,
         video_reader=video_reader,
+        trace_writer=trace_writer,
+        trace_context=trace_context,
     )
 
 

decode_content/readers/video.py

@@ -5,6 +5,7 @@ import base64
 import logging
 import os
 import re
+import time
 from pathlib import Path
 from typing import Any, Callable, Optional
 
@@ -15,6 +16,7 @@ from core.jsonio import extract_json_object
 from core.media_download import download_media_bytes
 from core.models import Card, CardExtract, ExtractedContent, Post
 from core.prompts import load_prompt
+from pipeline.tracing import TraceContext, TraceWriter, hash_prompt, redact_headers, timed_ms
 
 logger = logging.getLogger(__name__)
 
@@ -62,6 +64,8 @@ def extract_video(
     save_path: Optional[Path] = None,
     public_url: Optional[str] = None,
     oss_video_url: Optional[str] = None,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> ExtractedContent:
     key = settings.openrouter_api_key
     if not key:
@@ -111,18 +115,51 @@ def extract_video(
             }
         ],
     }
+    started = time.perf_counter()
+    headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
     try:
         resp = http_post(
             f"{settings.openrouter_base_url.rstrip('/')}/chat/completions",
-            headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
+            headers=headers,
             json=body,
             timeout=timeout,
         )
         resp.raise_for_status()
-        content = resp.json()["choices"][0]["message"]["content"]
+        response_json = resp.json()
+        content = response_json["choices"][0]["message"]["content"]
     except httpx.HTTPError as exc:
+        if trace_writer is not None:
+            trace_writer.llm_call(
+                context=trace_context or TraceContext(stage="decode", substage="read_video"),
+                stage="decode",
+                substage="read_video",
+                provider="openrouter",
+                model_name=settings.video_model,
+                endpoint=f"{settings.openrouter_base_url.rstrip('/')}/chat/completions",
+                prompt_name="extract_video",
+                prompt_hash=hash_prompt(prompt),
+                request_payload={"headers": redact_headers(headers), **body},
+                status="failed",
+                error_message=str(exc),
+                latency_ms=timed_ms(started),
+            )
         raise VideoExtractError(f"openrouter_http_error: {exc}") from exc
     except (KeyError, IndexError, TypeError, ValueError) as exc:
+        if trace_writer is not None:
+            trace_writer.llm_call(
+                context=trace_context or TraceContext(stage="decode", substage="read_video"),
+                stage="decode",
+                substage="read_video",
+                provider="openrouter",
+                model_name=settings.video_model,
+                endpoint=f"{settings.openrouter_base_url.rstrip('/')}/chat/completions",
+                prompt_name="extract_video",
+                prompt_hash=hash_prompt(prompt),
+                request_payload={"headers": redact_headers(headers), **body},
+                status="failed",
+                error_message=str(exc),
+                latency_ms=timed_ms(started),
+            )
         raise VideoExtractError(f"openrouter_response_invalid: {exc}") from exc
 
     obj = extract_json_object(content)
@@ -143,6 +180,26 @@ def extract_video(
         )
         card_extracts.append(CardExtract(index=idx, content=_seg_content(segment)))
     post.cards = cards
+    if trace_writer is not None:
+        trace_writer.llm_call(
+            context=trace_context or TraceContext(stage="decode", substage="read_video"),
+            stage="decode",
+            substage="read_video",
+            provider="openrouter",
+            model_name=settings.video_model,
+            endpoint=f"{settings.openrouter_base_url.rstrip('/')}/chat/completions",
+            prompt_name="extract_video",
+            prompt_hash=hash_prompt(prompt),
+            request_payload={"headers": redact_headers(headers), **body},
+            response_payload=response_json,
+            parsed_payload={
+                "segment_count": len(card_extracts),
+                "overall": obj.get("overall"),
+                "video_title": obj.get("video_title"),
+            },
+            status="done",
+            latency_ms=timed_ms(started),
+        )
     return ExtractedContent(
         text=str(obj.get("overall") or obj.get("video_title") or ""),
         cards=card_extracts,
@@ -160,6 +217,8 @@ def read_video(
     save_path: Path | None = None,
     public_url: str | None = None,
     oss_video_url: str | None = None,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> ExtractedContent:
     kwargs: dict[str, Any] = {
         "settings": settings,
@@ -168,6 +227,8 @@ def read_video(
         "save_path": save_path,
         "public_url": public_url or oss_video_url,
         "oss_video_url": oss_video_url,
+        "trace_writer": trace_writer,
+        "trace_context": trace_context,
     }
     if http_post is not None:
         kwargs["http_post"] = http_post

decode_content/repositories/postgres.py

@@ -357,23 +357,40 @@ class PostgresDecodeRepository:
         content_hash: str | None = None,
         source_path: str | None = None,
         snapshot: dict[str, Any] | None = None,
+        pipeline_run_id: UUID | None = None,
     ) -> ContractSnapshot:
         row = self._one(
             """
-            INSERT INTO contract_snapshots(
+            INSERT INTO contract_artifacts(
                 contract_name, contract_type, version_label, content_hash,
                 source_path, snapshot
             )
             VALUES (%s, %s, %s, %s, %s, %s)
+            ON CONFLICT (contract_name, contract_type, source_path, content_hash) DO UPDATE SET
+                version_label = COALESCE(EXCLUDED.version_label, contract_artifacts.version_label),
+                source_path = COALESCE(EXCLUDED.source_path, contract_artifacts.source_path),
+                snapshot = EXCLUDED.snapshot,
+                updated_at = now()
             RETURNING *
             """,
             (
                 contract_name,
                 contract_type,
                 version_label,
-                content_hash,
+                content_hash or "",
                 source_path,
                 _json(snapshot),
             ),
         )
+        if pipeline_run_id is not None:
+            self._one(
+                """
+                INSERT INTO run_contract_artifacts(pipeline_run_id, contract_artifact_id)
+                VALUES (%s, %s)
+                ON CONFLICT (pipeline_run_id, contract_artifact_id) DO UPDATE SET
+                    created_at = run_contract_artifacts.created_at
+                RETURNING *
+                """,
+                (pipeline_run_id, row["id"]),
+            )
         return ContractSnapshot.model_validate(row)

decode_content/repository.py

@@ -110,5 +110,6 @@ class DecodeRepository(Protocol):
         content_hash: str | None = None,
         source_path: str | None = None,
         snapshot: dict[str, Any] | None = None,
+        pipeline_run_id: UUID | None = None,
     ) -> ContractSnapshot:
         ...

decode_content/service.py

@@ -9,6 +9,7 @@ from uuid import UUID
 from core.config import Settings
 from core.llm import chat_json as default_chat_json
 from core.models import Post
+from pipeline.tracing import TraceContext, TraceWriter, hash_prompt, timed_ms
 from decode_content.contracts import SkillContract, load_contract
 from decode_content.framing import frame_and_clean
 from decode_content.gates import ChatJsonFn, creation_gate
@@ -17,6 +18,7 @@ from decode_content.payloads import build_payloads, validate_ingest_payload
 from decode_content.readers.service import read_post
 from decode_content.repository import DecodeRepository
 from decode_content.scoping import ScopeLinker, apply_scopes, nounify_scopes, scope_candidates
+import time
 
 
 @dataclass
@@ -48,6 +50,8 @@ class DecodeService:
         scope_linker: ScopeLinker | None = None,
         chat_json_fn: ChatJsonFn | None = None,
         reader: Callable[[Post], ReadResult] | None = None,
+        trace_writer: TraceWriter | None = None,
+        trace_context: TraceContext | None = None,
     ) -> None:
         self.settings = settings
         self.repository = repository
@@ -55,11 +59,69 @@ class DecodeService:
         self.scope_linker = scope_linker
         self.chat_json_fn = chat_json_fn or default_chat_json
         self.reader = reader
+        self.trace_writer = trace_writer
+        self.trace_context = trace_context or TraceContext(stage="decode")
 
-    def _read(self, post: Post) -> ReadResult:
+    def _read(self, post: Post, context: TraceContext | None = None) -> ReadResult:
         if self.reader is not None:
             return self.reader(post)
-        return read_post(post, settings=self.settings)
+        return read_post(
+            post,
+            settings=self.settings,
+            trace_writer=self.trace_writer,
+            trace_context=context or self.trace_context,
+        )
+
+    def _event(self, context: TraceContext, event_type: str, **kwargs: Any) -> None:
+        if self.trace_writer is None:
+            return
+        self.trace_writer.event(context=context, stage="decode", event_type=event_type, **kwargs)
+
+    def _traced_chat(self, context: TraceContext, substage: str) -> ChatJsonFn:
+        def call(system: str, user: str, **kwargs: Any) -> dict[str, Any]:
+            started = time.perf_counter()
+            try:
+                result = self.chat_json_fn(system, user, **kwargs)
+                if self.trace_writer is not None:
+                    self.trace_writer.llm_call(
+                        context=context.child(substage=substage),
+                        stage="decode",
+                        substage=substage,
+                        provider="bailian",
+                        model_name=self.settings.llm_model if self.settings else None,
+                        prompt_name=substage,
+                        prompt_hash=hash_prompt(system),
+                        request_payload={
+                            "system": system,
+                            "user": user,
+                            "timeout": kwargs.get("timeout"),
+                        },
+                        parsed_payload=result,
+                        status="done",
+                        latency_ms=timed_ms(started),
+                    )
+                return result
+            except Exception as exc:
+                if self.trace_writer is not None:
+                    self.trace_writer.llm_call(
+                        context=context.child(substage=substage),
+                        stage="decode",
+                        substage=substage,
+                        provider="bailian",
+                        model_name=self.settings.llm_model if self.settings else None,
+                        prompt_name=substage,
+                        prompt_hash=hash_prompt(system),
+                        request_payload={
+                            "system": system,
+                            "user": user,
+                            "timeout": kwargs.get("timeout"),
+                        },
+                        status="failed",
+                        error_message=str(exc),
+                        latency_ms=timed_ms(started),
+                    )
+                raise
+        return call
 
     def decode_post(
         self,
@@ -70,8 +132,26 @@ class DecodeService:
     ) -> DecodeWorkflowOutput:
         repo = self.repository
         job = repo.create_decode_job(item_id=item_id, status="running") if repo else None
-        self._save_contract_snapshots()
-        read = read_result or self._read(post)
+        context = self.trace_context.child(
+            item_id=item_id,
+            decode_job_id=getattr(job, "id", None),
+            stage="decode",
+        )
+        self._event(
+            context,
+            "decode_started",
+            status="running",
+            target_table="decode_jobs",
+            target_id=getattr(job, "id", None),
+        )
+        self._save_contract_artifacts()
+        read = read_result or self._read(post, context=context.child(substage="read_post"))
+        self._event(
+            context,
+            "read_completed",
+            status="done" if not read.is_empty else "skipped",
+            payload={"is_empty": read.is_empty, "card_count": len(read.cards or [])},
+        )
         if read.is_empty:
             gate = GateResult(passed=False, reason="读懂结果为空", details={"is_empty": True})
             decode_result = self._save_decode_result(
@@ -90,7 +170,13 @@ class DecodeService:
                 status="skipped",
             )
 
-        gate = creation_gate(read.text, chat_json_fn=self.chat_json_fn)
+        gate = creation_gate(read.text, chat_json_fn=self._traced_chat(context, "creation_gate"))
+        self._event(
+            context,
+            "creation_gate_completed",
+            status="done" if gate.passed else "skipped",
+            payload=gate.model_dump(mode="json"),
+        )
         if not gate.passed:
             decode_result = self._save_decode_result(
                 item_id=item_id,
@@ -112,12 +198,34 @@ class DecodeService:
             post,
             read.text,
             contract=self.contract,
-            chat_json_fn=self.chat_json_fn,
+            chat_json_fn=self._traced_chat(context, "frame_and_clean"),
+        )
+        self._event(
+            context,
+            "framing_completed",
+            status="done",
+            payload={"knowledge_count": len(knowledges)},
+        )
+        scopes = scope_candidates(
+            knowledges,
+            contract=self.contract,
+            chat_json_fn=self._traced_chat(context, "scope_candidates"),
         )
-        scopes = scope_candidates(knowledges, contract=self.contract, chat_json_fn=self.chat_json_fn)
-        scopes = nounify_scopes(scopes, chat_json_fn=self.chat_json_fn)
+        scopes = nounify_scopes(scopes, chat_json_fn=self._traced_chat(context, "scope_nounify"))
         apply_scopes(knowledges, scopes, self.scope_linker)
+        self._event(
+            context,
+            "scope_completed",
+            status="done",
+            payload={"scope_group_count": len(scopes)},
+        )
         payloads = build_payloads(post, knowledges)
+        self._event(
+            context,
+            "payloads_built",
+            status="done",
+            payload={"payload_count": len(payloads)},
+        )
         decode_result = self._save_decode_result(
             item_id=item_id,
             job_id=getattr(job, "id", None),
@@ -132,6 +240,14 @@ class DecodeService:
             knowledges=knowledges,
             payloads=payloads,
         )
+        self._event(
+            context.child(decode_result_id=getattr(decode_result, "id", None)),
+            "decode_finished",
+            status="decoded",
+            target_table="decode_results",
+            target_id=getattr(decode_result, "id", None),
+            payload={"knowledge_count": len(knowledges), "payload_draft_count": len(payload_drafts)},
+        )
         return DecodeWorkflowOutput(
             item_id=item_id,
             read_result=read,
@@ -143,7 +259,7 @@ class DecodeService:
             status="decoded",
         )
 
-    def _save_contract_snapshots(self) -> None:
+    def _save_contract_artifacts(self) -> None:
         if self.repository is None:
             return
         for snapshot in self.contract.snapshots():
@@ -154,6 +270,7 @@ class DecodeService:
                 content_hash=snapshot.content_hash,
                 source_path=snapshot.source_path,
                 snapshot=snapshot.snapshot,
+                pipeline_run_id=self.trace_context.pipeline_run_id,
             )
 
     def _save_decode_result(
@@ -257,6 +374,8 @@ def decode_post(
     chat_json_fn: ChatJsonFn | None = None,
     reader: Callable[[Post], ReadResult] | None = None,
     read_result: ReadResult | None = None,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> DecodeWorkflowOutput:
     service = DecodeService(
         settings=settings,
@@ -264,6 +383,8 @@ def decode_post(
         scope_linker=scope_linker,
         chat_json_fn=chat_json_fn,
         reader=reader,
+        trace_writer=trace_writer,
+        trace_context=trace_context,
     )
     return service.decode_post(item_id=item_id, post=post, read_result=read_result)
 

pipeline/__init__.py

@@ -1,6 +1,5 @@
 """Formal creation pipeline orchestration package."""
 
-from pipeline.creation_pipeline import CreationPipelineResult, run_creation_pipeline
 from pipeline.dedupe import dedupe_candidate_items, item_dedupe_key
 from pipeline.models import PipelineJob, PipelineRun
 
@@ -12,3 +11,15 @@ __all__ = [
     "item_dedupe_key",
     "run_creation_pipeline",
 ]
+
+
+def __getattr__(name: str):
+    if name in {"CreationPipelineResult", "run_creation_pipeline"}:
+        from pipeline.creation_pipeline import CreationPipelineResult, run_creation_pipeline
+
+        values = {
+            "CreationPipelineResult": CreationPipelineResult,
+            "run_creation_pipeline": run_creation_pipeline,
+        }
+        return values[name]
+    raise AttributeError(name)

pipeline/decode_runner.py

@@ -10,6 +10,7 @@ from core.text_limits import ERROR_MESSAGE_MAX_CHARS, clip_text
 from decode_content.readers.service import post_from_candidate_item
 from decode_content.service import DecodeService, DecodeWorkflowOutput
 from pipeline.dedupe import dedupe_candidate_items, should_decode_item
+from pipeline.tracing import NoopTraceWriter, TraceContext, TraceWriter
 
 
 class DecodeCandidateRepository(Protocol):
@@ -67,27 +68,93 @@ def run_decode_stage(
     run_id: UUID | None = None,
     limit: int = 100,
     decoded_item_ids: set[str] | None = None,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
 ) -> DecodeBatchResult:
     items = dedupe_candidate_items(candidate_repo.list_creation_candidate_items(run_id=run_id, limit=limit))
+    trace_writer = trace_writer or NoopTraceWriter()
+    base_context = trace_context or TraceContext(acquisition_run_id=run_id, stage="decode")
+    trace_writer.event(
+        context=base_context,
+        stage="decode",
+        event_type="decode_stage_started",
+        status="running",
+        payload={"run_id": str(run_id) if run_id else None, "candidate_count": len(items), "limit": limit},
+    )
     outputs: list[DecodeWorkflowOutput] = []
     failures: list[dict[str, str]] = []
     decoded = skipped = failed = 0
     for item in items:
         if item.id is None:
             skipped += 1
+            trace_writer.event(
+                context=base_context.child(platform=item.platform),
+                stage="decode",
+                event_type="decode_item_skipped",
+                status="skipped",
+                payload={"reason": "missing_item_id", "title": item.title or ""},
+            )
             continue
+        item_context = base_context.child(
+            item_id=item.id,
+            acquisition_job_id=item.job_id,
+            query_id=item.query_id,
+            platform=item.platform,
+        )
         decision = should_decode_item(item, decoded_item_ids=decoded_item_ids)
         if not decision.keep:
             skipped += 1
+            trace_writer.event(
+                context=item_context,
+                stage="decode",
+                event_type="decode_item_skipped",
+                status="skipped",
+                target_table="candidate_items",
+                target_id=item.id,
+                payload={"reason": decision.reason},
+            )
             continue
         try:
+            trace_writer.event(
+                context=item_context,
+                stage="decode",
+                event_type="decode_item_started",
+                status="running",
+                target_table="candidate_items",
+                target_id=item.id,
+            )
             media = candidate_repo.list_media_assets_for_item(item.id)
             post = post_from_candidate_item(item, media)
             outputs.append(decode_service.decode_post(item_id=item.id, post=post))
             decoded += 1
+            trace_writer.event(
+                context=item_context,
+                stage="decode",
+                event_type="decode_item_finished",
+                status="done",
+                target_table="candidate_items",
+                target_id=item.id,
+            )
         except Exception as exc:
             failed += 1
             failures.append(_record_failure(decode_service, item, exc))
+            trace_writer.event(
+                context=item_context,
+                stage="decode",
+                event_type="decode_item_failed",
+                status="failed",
+                severity="error",
+                target_table="candidate_items",
+                target_id=item.id,
+                error_message=str(exc),
+            )
+    trace_writer.event(
+        context=base_context,
+        stage="decode",
+        event_type="decode_stage_finished",
+        status="done" if failed == 0 else "partial",
+        payload={"total": len(items), "decoded": decoded, "skipped": skipped, "failed": failed},
+    )
     return DecodeBatchResult(
         total=len(items),
         decoded=decoded,

pipeline/models.py

@@ -50,9 +50,37 @@ class PipelineJob(PipelineModel):
     run_id: UUID | None = None
     stage: PipelineStage
     status: PipelineStatus = "pending"
+    target_table: str | None = None
     target_id: UUID | None = None
     attempt_count: int = 0
     metadata: dict[str, Any] = Field(default_factory=dict)
     error_message: str | None = None
     started_at: datetime | None = None
     finished_at: datetime | None = None
+
+
+class PipelineRunEvent(PipelineModel):
+    id: UUID | None = None
+    pipeline_run_id: UUID | None = None
+    pipeline_job_id: UUID | None = None
+    stage: PipelineStage | str
+    event_type: str
+    status: str | None = None
+    severity: str = "info"
+    target_table: str | None = None
+    target_id: UUID | None = None
+    acquisition_run_id: UUID | None = None
+    acquisition_job_id: UUID | None = None
+    query_id: UUID | None = None
+    item_id: UUID | None = None
+    decode_job_id: UUID | None = None
+    decode_result_id: UUID | None = None
+    payload_draft_id: UUID | None = None
+    ingest_record_id: UUID | None = None
+    platform: str | None = None
+    attempt_index: int | None = None
+    message: str | None = None
+    payload: dict[str, Any] = Field(default_factory=dict)
+    error_message: str | None = None
+    duration_ms: int | None = None
+    created_at: datetime | None = None

pipeline/postgres.py

@@ -0,0 +1,487 @@
+"""PostgreSQL repository and best-effort writer for pipeline tracing."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+from uuid import UUID
+
+import psycopg2.extras
+
+from core.config import CreationDbConfig
+from core.db_session import transaction
+from pipeline.models import PipelineJob, PipelineRun, PipelineRunEvent, PipelineStage
+from pipeline.tracing import TraceConfig, TraceContext, sanitize_payload
+
+Json = psycopg2.extras.Json
+psycopg2.extras.register_uuid()
+logger = logging.getLogger(__name__)
+
+
+class PostgresPipelineRepository:
+    """Repository for orchestration state and trace ledger rows.
+
+    This class owns no transaction boundary; API dependencies pass a pooled
+    connection, while the best-effort trace writer opens short transactions.
+    """
+
+    def __init__(self, conn: Any):
+        self.conn = conn
+
+    def _one(self, sql: str, params: tuple[Any, ...]) -> dict[str, Any]:
+        with self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute(sql, params)
+            row = cur.fetchone()
+            if row is None:
+                raise RuntimeError("expected one row, got none")
+            return dict(row)
+
+    def _one_or_none(self, sql: str, params: tuple[Any, ...]) -> dict[str, Any] | None:
+        with self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute(sql, params)
+            row = cur.fetchone()
+            return dict(row) if row else None
+
+    def _all(self, sql: str, params: tuple[Any, ...]) -> list[dict[str, Any]]:
+        with self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute(sql, params)
+            return [dict(row) for row in cur.fetchall()]
+
+    def create_pipeline_run(
+        self,
+        *,
+        run_key: str | None = None,
+        batch_id: UUID | None = None,
+        status: str = "pending",
+        current_stage: str | None = None,
+        config: dict[str, Any] | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> PipelineRun:
+        row = self._one(
+            """
+            INSERT INTO pipeline_runs(
+                run_key, batch_id, status, current_stage, config, metadata, started_at
+            )
+            VALUES (%s, %s, %s, %s, %s, %s, CASE WHEN %s = 'running' THEN now() ELSE NULL END)
+            ON CONFLICT (run_key) DO UPDATE SET
+                batch_id = COALESCE(EXCLUDED.batch_id, pipeline_runs.batch_id),
+                status = EXCLUDED.status,
+                current_stage = COALESCE(EXCLUDED.current_stage, pipeline_runs.current_stage),
+                config = pipeline_runs.config || EXCLUDED.config,
+                metadata = pipeline_runs.metadata || EXCLUDED.metadata,
+                started_at = COALESCE(pipeline_runs.started_at, EXCLUDED.started_at)
+            RETURNING *
+            """,
+            (
+                run_key,
+                batch_id,
+                status,
+                current_stage,
+                Json(config or {}),
+                Json(metadata or {}),
+                status,
+            ),
+        )
+        return _pipeline_run(row)
+
+    def mark_pipeline_run_status(
+        self,
+        run_id: UUID,
+        *,
+        status: str,
+        current_stage: str | None = None,
+        error_message: str | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> PipelineRun:
+        row = self._one(
+            """
+            UPDATE pipeline_runs SET
+                status = %s,
+                current_stage = COALESCE(%s, current_stage),
+                error_message = %s,
+                metadata = CASE WHEN %s THEN metadata || %s ELSE metadata END,
+                finished_at = CASE WHEN %s IN ('done', 'partial', 'failed') THEN now() ELSE finished_at END
+            WHERE id = %s
+            RETURNING *
+            """,
+            (
+                status,
+                current_stage,
+                error_message,
+                metadata is not None,
+                Json(metadata or {}),
+                status,
+                run_id,
+            ),
+        )
+        return _pipeline_run(row)
+
+    def get_pipeline_run(self, run_id: UUID) -> PipelineRun:
+        return _pipeline_run(self._one("SELECT * FROM pipeline_runs WHERE id = %s", (run_id,)))
+
+    def get_pipeline_run_by_acquisition_run(self, acquisition_run_id: UUID) -> PipelineRun:
+        row = self._one(
+            """
+            SELECT pr.* FROM pipeline_runs pr
+            JOIN pipeline_run_events pre ON pre.pipeline_run_id = pr.id
+            WHERE pre.acquisition_run_id = %s
+            ORDER BY pre.created_at DESC
+            LIMIT 1
+            """,
+            (acquisition_run_id,),
+        )
+        return _pipeline_run(row)
+
+    def save_pipeline_job(
+        self,
+        *,
+        run_id: UUID,
+        stage: PipelineStage | str,
+        target_id: UUID | None = None,
+        target_table: str | None = None,
+        status: str = "pending",
+        metadata: dict[str, Any] | None = None,
+    ) -> PipelineJob:
+        row = self._one(
+            """
+            INSERT INTO pipeline_jobs(
+                pipeline_run_id, stage, target_table, target_id, status, metadata, started_at
+            )
+            VALUES (%s, %s, %s, %s, %s, %s, CASE WHEN %s = 'running' THEN now() ELSE NULL END)
+            RETURNING *
+            """,
+            (run_id, stage, target_table, target_id, status, Json(metadata or {}), status),
+        )
+        return _pipeline_job(row)
+
+    def mark_job_status(
+        self,
+        job_id: UUID,
+        *,
+        status: str,
+        error_message: str | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> PipelineJob:
+        row = self._one(
+            """
+            UPDATE pipeline_jobs SET
+                status = %s,
+                error_message = %s,
+                metadata = CASE WHEN %s THEN metadata || %s ELSE metadata END,
+                finished_at = CASE WHEN %s IN ('done', 'partial', 'failed', 'skipped') THEN now() ELSE finished_at END
+            WHERE id = %s
+            RETURNING *
+            """,
+            (
+                status,
+                error_message,
+                metadata is not None,
+                Json(metadata or {}),
+                status,
+                job_id,
+            ),
+        )
+        return _pipeline_job(row)
+
+    def append_event(
+        self,
+        *,
+        context: TraceContext,
+        stage: str,
+        event_type: str,
+        status: str | None = None,
+        severity: str = "info",
+        target_table: str | None = None,
+        target_id: UUID | None = None,
+        message: str | None = None,
+        payload: dict[str, Any] | None = None,
+        error_message: str | None = None,
+        duration_ms: int | None = None,
+        attempt_index: int | None = None,
+    ) -> PipelineRunEvent:
+        row = self._one(
+            """
+            INSERT INTO pipeline_run_events(
+                pipeline_run_id, pipeline_job_id, stage, event_type, status, severity,
+                target_table, target_id, acquisition_run_id, acquisition_job_id,
+                query_id, item_id, decode_job_id, decode_result_id, payload_draft_id,
+                ingest_record_id, platform, attempt_index, message, payload,
+                error_message, duration_ms
+            )
+            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+            RETURNING *
+            """,
+            (
+                context.pipeline_run_id,
+                context.pipeline_job_id,
+                stage,
+                event_type,
+                status,
+                severity,
+                target_table,
+                target_id,
+                context.acquisition_run_id,
+                context.acquisition_job_id,
+                context.query_id,
+                context.item_id,
+                context.decode_job_id,
+                context.decode_result_id,
+                context.payload_draft_id,
+                context.ingest_record_id,
+                context.platform,
+                attempt_index,
+                message,
+                Json(payload or {}),
+                error_message,
+                duration_ms,
+            ),
+        )
+        return PipelineRunEvent.model_validate(_event_row(row))
+
+    def record_candidate_hit(
+        self,
+        *,
+        context: TraceContext,
+        item_id: UUID | None,
+        platform: str,
+        unique_key: str | None = None,
+        platform_item_id: str | None = None,
+        search_provider: str | None = None,
+        detail_provider: str | None = None,
+        attempt_index: int | None = None,
+        page_index: int | None = None,
+        page_rank: int | None = None,
+        candidate_rank: int | None = None,
+        source_cursor: str | None = None,
+        is_duplicate_hit: bool = False,
+        raw_candidate: dict[str, Any] | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        return self._one(
+            """
+            INSERT INTO candidate_item_hits(
+                pipeline_run_id, acquisition_run_id, acquisition_job_id,
+                query_id, item_id, platform, unique_key, platform_item_id,
+                search_provider, detail_provider, attempt_index, page_index,
+                page_rank, candidate_rank, source_cursor, is_duplicate_hit,
+                raw_candidate, metadata
+            )
+            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+            RETURNING *
+            """,
+            (
+                context.pipeline_run_id,
+                context.acquisition_run_id,
+                context.acquisition_job_id,
+                context.query_id,
+                item_id,
+                platform,
+                unique_key,
+                platform_item_id,
+                search_provider,
+                detail_provider,
+                attempt_index,
+                page_index,
+                page_rank,
+                candidate_rank,
+                source_cursor,
+                is_duplicate_hit,
+                Json(raw_candidate or {}),
+                Json(metadata or {}),
+            ),
+        )
+
+    def save_llm_call_trace(
+        self,
+        *,
+        context: TraceContext,
+        stage: str,
+        substage: str | None = None,
+        provider: str | None = None,
+        model_name: str | None = None,
+        endpoint: str | None = None,
+        prompt_name: str | None = None,
+        prompt_hash: str | None = None,
+        request_payload: dict[str, Any] | None = None,
+        response_payload: dict[str, Any] | None = None,
+        parsed_payload: dict[str, Any] | None = None,
+        status: str = "pending",
+        error_message: str | None = None,
+        latency_ms: int | None = None,
+        attempt_index: int | None = None,
+    ) -> dict[str, Any]:
+        return self._one(
+            """
+            INSERT INTO llm_call_traces(
+                pipeline_run_id, pipeline_job_id, stage, substage, provider,
+                model_name, endpoint, prompt_name, prompt_hash, trace_context,
+                request_payload, response_payload, parsed_payload, status,
+                error_message, latency_ms, attempt_index
+            )
+            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+            RETURNING *
+            """,
+            (
+                context.pipeline_run_id,
+                context.pipeline_job_id,
+                stage,
+                substage,
+                provider,
+                model_name,
+                endpoint,
+                prompt_name,
+                prompt_hash,
+                Json(context.as_payload()),
+                Json(request_payload or {}),
+                Json(response_payload or {}),
+                Json(parsed_payload or {}),
+                status,
+                error_message,
+                latency_ms,
+                attempt_index,
+            ),
+        )
+
+    def list_timeline(self, run_id: UUID) -> list[dict[str, Any]]:
+        return self._all(
+            """
+            SELECT * FROM pipeline_run_events
+            WHERE pipeline_run_id = %s
+            ORDER BY created_at, id
+            """,
+            (run_id,),
+        )
+
+    def list_jobs(self, run_id: UUID) -> list[dict[str, Any]]:
+        return self._all(
+            """
+            SELECT * FROM pipeline_jobs
+            WHERE pipeline_run_id = %s
+            ORDER BY created_at, id
+            """,
+            (run_id,),
+        )
+
+    def list_candidate_hits(self, run_id: UUID) -> list[dict[str, Any]]:
+        return self._all(
+            """
+            SELECT * FROM candidate_item_hits
+            WHERE pipeline_run_id = %s
+            ORDER BY created_at, page_index NULLS LAST, page_rank NULLS LAST, id
+            """,
+            (run_id,),
+        )
+
+    def list_llm_call_traces(self, run_id: UUID, *, limit: int = 500) -> list[dict[str, Any]]:
+        return self._all(
+            """
+            SELECT * FROM llm_call_traces
+            WHERE pipeline_run_id = %s
+            ORDER BY created_at, id
+            LIMIT %s
+            """,
+            (run_id, limit),
+        )
+
+    def get_timeline_bundle(self, run_id: UUID) -> dict[str, Any]:
+        return {
+            "events": self.list_timeline(run_id),
+            "jobs": self.list_jobs(run_id),
+            "candidate_hits": self.list_candidate_hits(run_id),
+            "llm_call_traces": self.list_llm_call_traces(run_id),
+        }
+
+    def get_resume_cursor(self, run_id: UUID):
+        return None
+
+
+class PostgresTraceWriter:
+    """Best-effort writer that never lets tracing break the business flow."""
+
+    def __init__(self, *, db_config: CreationDbConfig, config: TraceConfig):
+        self.db_config = db_config
+        self.config = config
+
+    def event(self, *, context: TraceContext, stage: str, event_type: str, **kwargs: Any) -> None:
+        self._best_effort(
+            lambda repo: repo.append_event(
+                context=context,
+                stage=stage,
+                event_type=event_type,
+                payload=sanitize_payload(kwargs.pop("payload", {}) or {}, max_chars=self.config.max_payload_chars),
+                **kwargs,
+            )
+        )
+
+    def candidate_hit(self, *, context: TraceContext, **kwargs: Any) -> None:
+        self._best_effort(
+            lambda repo: repo.record_candidate_hit(
+                context=context,
+                raw_candidate=sanitize_payload(kwargs.pop("raw_candidate", {}) or {}, max_chars=self.config.max_payload_chars),
+                metadata=sanitize_payload(kwargs.pop("metadata", {}) or {}, max_chars=self.config.max_payload_chars),
+                **kwargs,
+            )
+        )
+
+    def llm_call(self, *, context: TraceContext, stage: str, substage: str | None = None, **kwargs: Any) -> None:
+        request_payload = kwargs.pop("request_payload", {}) if self.config.capture_request else {}
+        response_payload = kwargs.pop("response_payload", {}) if self.config.capture_response else {}
+        parsed_payload = kwargs.pop("parsed_payload", {})
+        self._best_effort(
+            lambda repo: repo.save_llm_call_trace(
+                context=context,
+                stage=stage,
+                substage=substage,
+                request_payload=sanitize_payload(request_payload or {}, max_chars=self.config.max_payload_chars),
+                response_payload=sanitize_payload(response_payload or {}, max_chars=self.config.max_payload_chars),
+                parsed_payload=sanitize_payload(parsed_payload or {}, max_chars=self.config.max_payload_chars),
+                **kwargs,
+            )
+        )
+
+    def _best_effort(self, fn) -> None:
+        if not self.config.enabled:
+            return
+        try:
+            with transaction(self.db_config) as conn:
+                fn(PostgresPipelineRepository(conn))
+        except Exception as exc:  # pragma: no cover - defensive isolation
+            logger.warning("pipeline trace write skipped: %s", exc)
+
+
+def _pipeline_run(row: dict[str, Any]) -> PipelineRun:
+    return PipelineRun(
+        id=row.get("id"),
+        run_key=row.get("run_key"),
+        batch_id=row.get("batch_id"),
+        status=row.get("status") or "pending",
+        current_stage=row.get("current_stage"),
+        metadata={
+            **(row.get("config") or {}),
+            **(row.get("metadata") or {}),
+        },
+        error_message=row.get("error_message"),
+        started_at=row.get("started_at"),
+        finished_at=row.get("finished_at"),
+    )
+
+
+def _pipeline_job(row: dict[str, Any]) -> PipelineJob:
+    return PipelineJob(
+        id=row.get("id"),
+        run_id=row.get("pipeline_run_id"),
+        stage=row.get("stage"),
+        status=row.get("status") or "pending",
+        target_table=row.get("target_table"),
+        target_id=row.get("target_id"),
+        attempt_count=row.get("attempt_count") or 0,
+        metadata=row.get("metadata") or {},
+        error_message=row.get("error_message"),
+        started_at=row.get("started_at"),
+        finished_at=row.get("finished_at"),
+    )
+
+
+def _event_row(row: dict[str, Any]) -> dict[str, Any]:
+    out = dict(row)
+    out["pipeline_run_id"] = out.pop("pipeline_run_id", None)
+    return out

pipeline/tracing.py

@@ -0,0 +1,190 @@
+"""Best-effort tracing helpers for the formal creation pipeline."""
+from __future__ import annotations
+
+import hashlib
+import logging
+import os
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Protocol
+from uuid import UUID
+
+from core.config import CreationDbConfig, env_value, load_env_file
+from core.text_limits import clip_text
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class TraceConfig:
+    enabled: bool = True
+    capture_request: bool = True
+    capture_response: bool = True
+    max_payload_chars: int = 50_000
+
+    @classmethod
+    def from_env(cls, env_file: str | Path = ".env") -> "TraceConfig":
+        file_env = load_env_file(env_file)
+        return cls(
+            enabled=_env_bool(env_value("CK_TRACE_ENABLED", file_env, "true")),
+            capture_request=_env_bool(env_value("CK_TRACE_CAPTURE_REQUEST", file_env, "true")),
+            capture_response=_env_bool(env_value("CK_TRACE_CAPTURE_RESPONSE", file_env, "true")),
+            max_payload_chars=_env_int(env_value("CK_TRACE_MAX_PAYLOAD_CHARS", file_env, "50000"), 50_000),
+        )
+
+
+@dataclass(frozen=True)
+class TraceContext:
+    pipeline_run_id: UUID | None = None
+    pipeline_job_id: UUID | None = None
+    acquisition_run_id: UUID | None = None
+    acquisition_job_id: UUID | None = None
+    query_id: UUID | None = None
+    item_id: UUID | None = None
+    decode_job_id: UUID | None = None
+    decode_result_id: UUID | None = None
+    payload_draft_id: UUID | None = None
+    ingest_record_id: UUID | None = None
+    platform: str | None = None
+    stage: str | None = None
+    substage: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    def child(self, **updates: Any) -> "TraceContext":
+        values = {
+            "pipeline_run_id": self.pipeline_run_id,
+            "pipeline_job_id": self.pipeline_job_id,
+            "acquisition_run_id": self.acquisition_run_id,
+            "acquisition_job_id": self.acquisition_job_id,
+            "query_id": self.query_id,
+            "item_id": self.item_id,
+            "decode_job_id": self.decode_job_id,
+            "decode_result_id": self.decode_result_id,
+            "payload_draft_id": self.payload_draft_id,
+            "ingest_record_id": self.ingest_record_id,
+            "platform": self.platform,
+            "stage": self.stage,
+            "substage": self.substage,
+            "metadata": dict(self.metadata),
+        }
+        metadata = updates.pop("metadata", None)
+        values.update(updates)
+        if metadata:
+            values["metadata"].update(metadata)
+        return TraceContext(**values)
+
+    def as_payload(self) -> dict[str, Any]:
+        out: dict[str, Any] = {
+            "pipeline_run_id": self.pipeline_run_id,
+            "pipeline_job_id": self.pipeline_job_id,
+            "acquisition_run_id": self.acquisition_run_id,
+            "acquisition_job_id": self.acquisition_job_id,
+            "query_id": self.query_id,
+            "item_id": self.item_id,
+            "decode_job_id": self.decode_job_id,
+            "decode_result_id": self.decode_result_id,
+            "payload_draft_id": self.payload_draft_id,
+            "ingest_record_id": self.ingest_record_id,
+            "platform": self.platform,
+            "stage": self.stage,
+            "substage": self.substage,
+            **self.metadata,
+        }
+        return {key: str(value) if isinstance(value, UUID) else value for key, value in out.items() if value is not None}
+
+
+class TraceWriter(Protocol):
+    def event(self, *, context: TraceContext, stage: str, event_type: str, **kwargs: Any) -> None:
+        ...
+
+    def candidate_hit(self, *, context: TraceContext, **kwargs: Any) -> None:
+        ...
+
+    def llm_call(self, *, context: TraceContext, stage: str, substage: str | None = None, **kwargs: Any) -> None:
+        ...
+
+
+class NoopTraceWriter:
+    def event(self, *, context: TraceContext, stage: str, event_type: str, **kwargs: Any) -> None:
+        return None
+
+    def candidate_hit(self, *, context: TraceContext, **kwargs: Any) -> None:
+        return None
+
+    def llm_call(self, *, context: TraceContext, stage: str, substage: str | None = None, **kwargs: Any) -> None:
+        return None
+
+
+def new_trace_writer(db_config: CreationDbConfig, *, env_file: str | Path = ".env") -> TraceWriter:
+    config = TraceConfig.from_env(env_file)
+    if not config.enabled:
+        return NoopTraceWriter()
+    from pipeline.postgres import PostgresTraceWriter
+
+    return PostgresTraceWriter(db_config=db_config, config=config)
+
+
+def hash_prompt(value: str | None) -> str | None:
+    if not value:
+        return None
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def redact_headers(headers: dict[str, Any] | None) -> dict[str, Any]:
+    if not headers:
+        return {}
+    out = dict(headers)
+    for key in list(out):
+        if key.lower() in {"authorization", "api-key", "x-api-key"}:
+            out[key] = "<redacted>"
+    return out
+
+
+def sanitize_payload(value: Any, *, max_chars: int) -> Any:
+    if value is None or isinstance(value, (bool, int, float)):
+        return value
+    if isinstance(value, UUID):
+        return str(value)
+    if isinstance(value, str):
+        if value.startswith("data:") and ";base64," in value[:80]:
+            head, _, data = value.partition(",")
+            digest = hashlib.sha256(data.encode("utf-8")).hexdigest() if data else ""
+            return {
+                "omitted": "base64_data_url",
+                "media_type": head[:80],
+                "chars": len(value),
+                "sha256": digest,
+            }
+        return clip_text(value, max_chars)
+    if isinstance(value, dict):
+        return {
+            str(key): sanitize_payload(item, max_chars=max_chars)
+            for key, item in value.items()
+        }
+    if isinstance(value, (list, tuple)):
+        return [sanitize_payload(item, max_chars=max_chars) for item in value]
+    if hasattr(value, "model_dump"):
+        return sanitize_payload(value.model_dump(mode="json"), max_chars=max_chars)
+    return clip_text(str(value), max_chars)
+
+
+def timed_ms(start: float) -> int:
+    return max(0, int((time.perf_counter() - start) * 1000))
+
+
+def _env_bool(value: str) -> bool:
+    return value.strip().lower() not in {"0", "false", "no", "off"}
+
+
+def _env_int(value: str, default: int) -> int:
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        logger.warning("invalid trace integer config %r; using %s", value, default)
+        return default
+    return parsed if parsed > 0 else default
+
+
+def current_env_file(default: str = ".env") -> str:
+    return os.getenv("CK_ENV_FILE", default)

scripts/backfill_pipeline_traces.py

@@ -0,0 +1,184 @@
+#!/usr/bin/env python3
+"""Backfill pipeline trace ledger rows from existing acquisition runs.
+
+Dry-run is the default. This script never fabricates historical LLM traces:
+old runs only get structural events plus an explicit backfill note.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from dataclasses import asdict, is_dataclass
+from pathlib import Path
+from uuid import UUID
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from core.config import CreationDbConfig  # noqa: E402
+from core.db_session import transaction  # noqa: E402
+from pipeline.postgres import PostgresPipelineRepository  # noqa: E402
+from pipeline.tracing import TraceContext  # noqa: E402
+
+
+def _model_dump(value):
+    if hasattr(value, "model_dump"):
+        return value.model_dump(mode="json")
+    if is_dataclass(value):
+        return asdict(value)
+    if isinstance(value, dict):
+        return value
+    return dict(value)
+
+
+def _acquisition_runs(conn, *, limit: int | None = None) -> list[dict]:
+    sql = """
+        SELECT ar.*,
+               COUNT(DISTINCT aj.id)::int AS job_count,
+               COUNT(DISTINCT ci.id)::int AS candidate_count
+        FROM acquisition_runs ar
+        LEFT JOIN acquisition_jobs aj ON aj.run_id = ar.id
+        LEFT JOIN candidate_items ci ON ci.job_id = aj.id
+        GROUP BY ar.id
+        ORDER BY ar.created_at
+    """
+    if limit is not None:
+        sql += " LIMIT %s"
+        params = (limit,)
+    else:
+        params = ()
+    return PostgresPipelineRepository(conn)._all(sql, params)
+
+
+def _candidate_hits(conn, acquisition_run_id: UUID) -> list[dict]:
+    return PostgresPipelineRepository(conn)._all(
+        """
+        SELECT
+            aj.id AS acquisition_job_id,
+            aj.query_id,
+            aj.platform,
+            ci.id AS item_id,
+            ci.unique_key,
+            ci.platform_item_id,
+            ci.metadata,
+            ci.source_payload
+        FROM acquisition_jobs aj
+        JOIN candidate_items ci ON ci.job_id = aj.id
+        WHERE aj.run_id = %s
+        ORDER BY aj.platform, ci.created_at
+        """,
+        (acquisition_run_id,),
+    )
+
+
+def backfill(*, env_file: str, apply: bool, limit: int | None = None) -> dict:
+    db_config = CreationDbConfig.from_env(env_file)
+    with transaction(db_config) as conn:
+        repo = PostgresPipelineRepository(conn)
+        runs = _acquisition_runs(conn, limit=limit)
+        if not apply:
+            return {
+                "dry_run": True,
+                "acquisition_run_count": len(runs),
+                "candidate_hit_count": sum(int(row.get("candidate_count") or 0) for row in runs),
+                "runs": [
+                    {
+                        "acquisition_run_id": str(row["id"]),
+                        "run_key": row.get("run_key"),
+                        "status": row.get("status"),
+                        "job_count": row.get("job_count"),
+                        "candidate_count": row.get("candidate_count"),
+                    }
+                    for row in runs
+                ],
+            }
+
+        pipeline_count = 0
+        hit_count = 0
+        for row in runs:
+            acquisition_run_id = row["id"]
+            run = repo.create_pipeline_run(
+                run_key=f"backfill:{acquisition_run_id}",
+                batch_id=row.get("batch_id"),
+                status=row.get("status") or "done",
+                current_stage="ingest",
+                metadata={
+                    "source": "backfill_pipeline_traces",
+                    "acquisition_run_id": str(acquisition_run_id),
+                    "original_run_key": row.get("run_key"),
+                },
+            )
+            pipeline_count += 1
+            context = TraceContext(
+                pipeline_run_id=run.id,
+                acquisition_run_id=acquisition_run_id,
+                stage="search",
+            )
+            repo.append_event(
+                context=context,
+                stage="query",
+                event_type="backfilled_without_llm_trace",
+                status="done",
+                target_table="acquisition_runs",
+                target_id=acquisition_run_id,
+                payload={
+                    "reason": "historical raw LLM requests and responses were not stored",
+                    "job_count": row.get("job_count"),
+                    "candidate_count": row.get("candidate_count"),
+                },
+            )
+            for hit in _candidate_hits(conn, acquisition_run_id):
+                metadata = hit.get("metadata") or {}
+                repo.record_candidate_hit(
+                    context=context.child(
+                        acquisition_job_id=hit.get("acquisition_job_id"),
+                        query_id=hit.get("query_id"),
+                        item_id=hit.get("item_id"),
+                        platform=hit.get("platform"),
+                    ),
+                    item_id=hit.get("item_id"),
+                    platform=hit.get("platform"),
+                    unique_key=hit.get("unique_key"),
+                    platform_item_id=hit.get("platform_item_id"),
+                    search_provider=metadata.get("search_provider"),
+                    detail_provider=metadata.get("detail_provider"),
+                    attempt_index=metadata.get("retry_attempt_index"),
+                    page_index=metadata.get("page_index"),
+                    page_rank=metadata.get("page_rank"),
+                    candidate_rank=metadata.get("candidate_rank"),
+                    source_cursor=metadata.get("source_cursor"),
+                    is_duplicate_hit=metadata.get("acquisition_match_status") == "existing",
+                    raw_candidate=metadata.get("matched_candidate") or {},
+                    metadata={"source": "backfill_pipeline_traces"},
+                )
+                hit_count += 1
+        return {
+            "dry_run": False,
+            "pipeline_run_count": pipeline_count,
+            "candidate_hit_count": hit_count,
+        }
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--env-file", default=".env")
+    parser.add_argument("--apply", action="store_true", help="Actually write trace rows. Default is dry-run.")
+    parser.add_argument("--limit", type=int, default=None)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    print(json.dumps(
+        backfill(env_file=args.env_file, apply=args.apply, limit=args.limit),
+        ensure_ascii=False,
+        default=str,
+        indent=2,
+    ))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/ingest_payloads.py

@@ -0,0 +1,228 @@
+#!/usr/bin/env python3
+"""Ingest existing payload drafts for one acquisition run.
+
+This script does not search, classify, or decode. It only reads payload_drafts
+already produced for candidate items under the given run_id.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from dataclasses import asdict, is_dataclass
+from pathlib import Path
+from uuid import UUID
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from core.config import CreationDbConfig, IngestApiConfig  # noqa: E402
+from core.db_session import transaction  # noqa: E402
+from decode_content.ingest import KnowledgeIngestClient, ingest_payload_draft  # noqa: E402
+from decode_content.models import PayloadDraft  # noqa: E402
+from decode_content.repositories.postgres import PostgresDecodeRepository  # noqa: E402
+from pipeline.postgres import PostgresPipelineRepository  # noqa: E402
+from pipeline.tracing import TraceContext, new_trace_writer  # noqa: E402
+
+
+def _model_dump(value):
+    if hasattr(value, "model_dump"):
+        return value.model_dump(mode="json")
+    if is_dataclass(value):
+        return asdict(value)
+    if isinstance(value, dict):
+        return value
+    return dict(value)
+
+
+def list_payload_drafts_for_run(
+    repo: PostgresDecodeRepository,
+    *,
+    run_id: UUID,
+    include_real_ingested: bool = False,
+    limit: int | None = None,
+) -> list[PayloadDraft]:
+    skip_real_ingested = "" if include_real_ingested else """
+      AND NOT EXISTS (
+        SELECT 1 FROM ingest_records ir
+        WHERE ir.payload_draft_id = pd.id
+          AND ir.target_system = 'knowledge-ingest-api'
+          AND ir.status = 'ingested'
+      )
+    """
+    limit_sql = "LIMIT %s" if limit is not None else ""
+    params: tuple = (run_id, limit) if limit is not None else (run_id,)
+    rows = repo._all(
+        f"""
+        SELECT pd.*
+        FROM payload_drafts pd
+        JOIN candidate_items ci ON ci.id = pd.item_id
+        JOIN acquisition_jobs aj ON aj.id = ci.job_id
+        WHERE aj.run_id = %s
+        {skip_real_ingested}
+        ORDER BY ci.created_at, pd.created_at, pd.id
+        {limit_sql}
+        """,
+        params,
+    )
+    return [PayloadDraft.model_validate(row) for row in rows]
+
+
+def count_run_payloads(repo: PostgresDecodeRepository, *, run_id: UUID) -> dict[str, int]:
+    row = repo._one(
+        """
+        SELECT
+          COUNT(DISTINCT pd.id)::int AS total_payloads,
+          COUNT(DISTINCT pd.id) FILTER (
+            WHERE EXISTS (
+              SELECT 1 FROM ingest_records ir
+              WHERE ir.payload_draft_id = pd.id
+                AND ir.target_system = 'knowledge-ingest-api'
+                AND ir.status = 'ingested'
+            )
+          )::int AS already_real_ingested
+        FROM payload_drafts pd
+        JOIN candidate_items ci ON ci.id = pd.item_id
+        JOIN acquisition_jobs aj ON aj.id = ci.job_id
+        WHERE aj.run_id = %s
+        """,
+        (run_id,),
+    )
+    return {
+        "total_payloads": int(row.get("total_payloads") or 0),
+        "already_real_ingested": int(row.get("already_real_ingested") or 0),
+    }
+
+
+def ingest_existing_payloads(
+    repo: PostgresDecodeRepository,
+    drafts: list[PayloadDraft],
+    *,
+    dry_run: bool,
+    env_file: str,
+    trace_writer=None,
+    trace_context: TraceContext | None = None,
+) -> list[dict]:
+    client = None
+    if not dry_run:
+        config = IngestApiConfig.from_env(env_file)
+        if not config.url:
+            raise RuntimeError("CK_INGEST_API_URL 未配置，真实 ingest 已关闭")
+        client = KnowledgeIngestClient(config)
+
+    records: list[dict] = []
+    for draft in drafts:
+        if draft.id is None:
+            continue
+        context = (trace_context or TraceContext(stage="ingest")).child(
+            item_id=draft.item_id,
+            payload_draft_id=draft.id,
+            stage="ingest",
+        )
+        if trace_writer is not None:
+            trace_writer.event(
+                context=context,
+                stage="ingest",
+                event_type="ingest_started",
+                status="running",
+                target_table="payload_drafts",
+                target_id=draft.id,
+                payload={"dry_run": dry_run, "entrypoint": "scripts/ingest_payloads.py"},
+            )
+        record = ingest_payload_draft(repo, draft, dry_run=dry_run, client=client)
+        if trace_writer is not None:
+            trace_writer.event(
+                context=context.child(ingest_record_id=record.id),
+                stage="ingest",
+                event_type="ingest_recorded",
+                status=record.status,
+                target_table="ingest_records",
+                target_id=record.id,
+                payload={"dry_run": dry_run, "target_system": record.target_system},
+                error_message=record.error_message,
+            )
+        records.append(_model_dump(record))
+    return records
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--run-id", required=True, help="acquisition_runs.id whose existing payloads should be ingested")
+    parser.add_argument("--pipeline-run-id", help="Optional pipeline_runs.id for trace events")
+    parser.add_argument("--env-file", default=".env")
+    parser.add_argument("--real", action="store_true", help="Call CK_INGEST_API_URL. Default is dry-run.")
+    parser.add_argument("--limit", type=int, default=None, help="Optional max payload drafts to process")
+    parser.add_argument(
+        "--include-real-ingested",
+        action="store_true",
+        help="Also process drafts that already have successful knowledge-ingest-api records.",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    run_id = UUID(args.run_id)
+    db_config = CreationDbConfig.from_env(args.env_file)
+    pipeline_run_id = UUID(args.pipeline_run_id) if args.pipeline_run_id else None
+    if pipeline_run_id is None:
+        try:
+            with transaction(db_config) as conn:
+                pipeline_run_id = PostgresPipelineRepository(conn).get_pipeline_run_by_acquisition_run(run_id).id
+        except Exception:
+            pipeline_run_id = None
+    trace_writer = new_trace_writer(db_config, env_file=args.env_file)
+    trace_context = TraceContext(
+        pipeline_run_id=pipeline_run_id,
+        acquisition_run_id=run_id,
+        stage="ingest",
+    )
+
+    with transaction(db_config) as conn:
+        repo = PostgresDecodeRepository(conn)
+        counts = count_run_payloads(repo, run_id=run_id)
+        drafts = list_payload_drafts_for_run(
+            repo,
+            run_id=run_id,
+            include_real_ingested=args.include_real_ingested,
+            limit=args.limit,
+        )
+        records = ingest_existing_payloads(
+            repo,
+            drafts,
+            dry_run=not args.real,
+            env_file=args.env_file,
+            trace_writer=trace_writer,
+            trace_context=trace_context,
+        )
+
+    ingested_count = sum(1 for record in records if record.get("status") == "ingested")
+    failed_count = sum(1 for record in records if record.get("status") == "failed")
+    skipped_already_real = (
+        0
+        if args.include_real_ingested
+        else max(0, counts["already_real_ingested"])
+    )
+    print(json.dumps(
+        {
+            "run_id": str(run_id),
+            "pipeline_run_id": str(pipeline_run_id) if pipeline_run_id else None,
+            "dry_run": not args.real,
+            "total_payloads": counts["total_payloads"],
+            "already_real_ingested": counts["already_real_ingested"],
+            "selected_payloads": len(drafts),
+            "skipped_already_real_ingested": skipped_already_real,
+            "ingested_count": ingested_count,
+            "failed_count": failed_count,
+            "records": records,
+        },
+        ensure_ascii=False,
+        default=str,
+        indent=2,
+    ))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_creation_pipeline.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import logging
 from dataclasses import asdict, is_dataclass
 from uuid import UUID
 
@@ -15,6 +16,10 @@ from decode_content.ingest import KnowledgeIngestClient, ingest_payload_draft
 from decode_content.repositories.postgres import PostgresDecodeRepository
 from decode_content.service import DecodeService
 from pipeline.decode_runner import run_decode_stage
+from pipeline.postgres import PostgresPipelineRepository
+from pipeline.tracing import TraceContext, TraceWriter, new_trace_writer
+
+logger = logging.getLogger(__name__)
 
 
 def _model_dump(value):
@@ -27,18 +32,156 @@ def _model_dump(value):
     return dict(value)
 
 
-def _ingest_payloads(repo: PostgresDecodeRepository, outputs: list, *, dry_run: bool, env_file: str) -> list[dict]:
+def _ingest_payloads(
+    repo: PostgresDecodeRepository,
+    outputs: list,
+    *,
+    dry_run: bool,
+    env_file: str,
+    trace_writer: TraceWriter | None = None,
+    trace_context: TraceContext | None = None,
+) -> list[dict]:
     records: list[dict] = []
     client = None if dry_run else KnowledgeIngestClient(IngestApiConfig.from_env(env_file))
     for output in outputs:
         for draft in output.payload_drafts:
             if draft.id is None:
                 continue
+            context = (trace_context or TraceContext()).child(
+                item_id=output.item_id,
+                payload_draft_id=draft.id,
+                stage="ingest",
+            )
+            if trace_writer:
+                trace_writer.event(
+                    context=context,
+                    stage="ingest",
+                    event_type="ingest_started",
+                    target_table="payload_drafts",
+                    target_id=draft.id,
+                    payload={"dry_run": dry_run},
+                )
             record = ingest_payload_draft(repo, draft, dry_run=dry_run, client=client)
+            if trace_writer:
+                trace_writer.event(
+                    context=context.child(ingest_record_id=record.id),
+                    stage="ingest",
+                    event_type="ingest_recorded",
+                    status=record.status,
+                    target_table="ingest_records",
+                    target_id=record.id,
+                    payload={"dry_run": dry_run, "target_system": record.target_system},
+                    error_message=record.error_message,
+                )
             records.append(_model_dump(record))
     return records
 
 
+def _ensure_pipeline_run(
+    db_config: CreationDbConfig,
+    *,
+    batch_id: UUID,
+    run_key: str | None,
+    platforms: tuple[str, ...],
+    args: argparse.Namespace,
+) -> UUID | None:
+    try:
+        with transaction(db_config) as conn:
+            run = PostgresPipelineRepository(conn).create_pipeline_run(
+                run_key=run_key or f"creation-pipeline:{batch_id}",
+                batch_id=batch_id,
+                status="running",
+                current_stage="search",
+                config={
+                    "platforms": list(platforms),
+                    "search_limit": args.search_limit,
+                    "display_limit": args.display_limit,
+                    "decode_limit": args.decode_limit,
+                    "resume": not args.no_resume,
+                    "skip_done": not args.no_skip_done,
+                    "real_ingest": args.real_ingest,
+                },
+                metadata={"entrypoint": "scripts/run_creation_pipeline.py"},
+            )
+            return run.id
+    except Exception as exc:
+        logger.warning("pipeline run trace setup skipped: %s", exc)
+        return None
+
+
+def _mark_pipeline_run(
+    db_config: CreationDbConfig,
+    run_id: UUID | None,
+    *,
+    status: str,
+    current_stage: str | None = None,
+    error_message: str | None = None,
+    metadata: dict | None = None,
+) -> None:
+    if run_id is None:
+        return
+    try:
+        with transaction(db_config) as conn:
+            PostgresPipelineRepository(conn).mark_pipeline_run_status(
+                run_id,
+                status=status,
+                current_stage=current_stage,
+                error_message=error_message,
+                metadata=metadata,
+            )
+    except Exception as exc:
+        logger.warning("pipeline run trace status skipped: %s", exc)
+
+
+def _start_pipeline_job(
+    db_config: CreationDbConfig,
+    pipeline_run_id: UUID | None,
+    *,
+    stage: str,
+    target_table: str | None = None,
+    target_id: UUID | None = None,
+    metadata: dict | None = None,
+) -> UUID | None:
+    if pipeline_run_id is None:
+        return None
+    try:
+        with transaction(db_config) as conn:
+            job = PostgresPipelineRepository(conn).save_pipeline_job(
+                run_id=pipeline_run_id,
+                stage=stage,
+                target_table=target_table,
+                target_id=target_id,
+                status="running",
+                metadata=metadata,
+            )
+            return job.id
+    except Exception as exc:
+        logger.warning("pipeline job trace setup skipped: %s", exc)
+        return None
+
+
+def _mark_pipeline_job(
+    db_config: CreationDbConfig,
+    job_id: UUID | None,
+    *,
+    status: str,
+    error_message: str | None = None,
+    metadata: dict | None = None,
+) -> None:
+    if job_id is None:
+        return
+    try:
+        with transaction(db_config) as conn:
+            PostgresPipelineRepository(conn).mark_job_status(
+                job_id,
+                status=status,
+                error_message=error_message,
+                metadata=metadata,
+            )
+    except Exception as exc:
+        logger.warning("pipeline job trace status skipped: %s", exc)
+
+
 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--batch-id", required=True, help="Formal query batch UUID")
@@ -66,42 +209,158 @@ def main(argv: list[str] | None = None) -> int:
     db_config = CreationDbConfig.from_env(args.env_file)
     platforms = tuple(args.platform or DEFAULT_PLATFORMS)
     batch_id = UUID(args.batch_id)
+    pipeline_run_id = _ensure_pipeline_run(
+        db_config,
+        batch_id=batch_id,
+        run_key=args.run_key,
+        platforms=platforms,
+        args=args,
+    )
+    trace_writer = new_trace_writer(db_config, env_file=args.env_file)
+    search_job_id = _start_pipeline_job(
+        db_config,
+        pipeline_run_id,
+        stage="search",
+        target_table="query_batches",
+        target_id=batch_id,
+        metadata={"platforms": list(platforms)},
+    )
+    trace_context = TraceContext(pipeline_run_id=pipeline_run_id, pipeline_job_id=search_job_id, stage="search")
 
-    with transaction(db_config) as conn:
-        acquisition_repo = PostgresAcquisitionRepository(conn)
-        acquisition = run_batch(
-            acquisition_repo,
-            batch_id=batch_id,
-            settings=settings,
-            platforms=platforms,
-            search_limit=args.search_limit,
-            display_limit=args.display_limit,
-            classify=True,
-            resume=not args.no_resume,
-            skip_done=not args.no_skip_done,
-            run_key=args.run_key,
+    try:
+        trace_writer.event(
+            context=trace_context,
+            stage="search",
+            event_type="pipeline_started",
+            status="running",
+            payload={"batch_id": str(batch_id), "platforms": list(platforms)},
+        )
+        with transaction(db_config) as conn:
+            acquisition_repo = PostgresAcquisitionRepository(conn)
+            acquisition = run_batch(
+                acquisition_repo,
+                batch_id=batch_id,
+                settings=settings,
+                platforms=platforms,
+                search_limit=args.search_limit,
+                display_limit=args.display_limit,
+                classify=True,
+                resume=not args.no_resume,
+                skip_done=not args.no_skip_done,
+                run_key=args.run_key,
+                trace_writer=trace_writer,
+                trace_context=trace_context,
+            )
+        _mark_pipeline_job(
+            db_config,
+            search_job_id,
+            status="partial" if acquisition.failed else "done",
+            metadata={"acquisition_run_id": str(acquisition.run_id), "failed": acquisition.failed},
         )
 
-    with transaction(db_config) as conn:
-        acquisition_repo = PostgresAcquisitionRepository(conn)
-        decode_repo = PostgresDecodeRepository(conn)
-        decode_service = DecodeService(settings=settings, repository=decode_repo)
-        decode = run_decode_stage(
-            candidate_repo=acquisition_repo,
-            decode_service=decode_service,
-            run_id=acquisition.run_id,
-            limit=args.decode_limit,
+        decode_job_id = _start_pipeline_job(
+            db_config,
+            pipeline_run_id,
+            stage="decode",
+            target_table="acquisition_runs",
+            target_id=acquisition.run_id,
+            metadata={"decode_limit": args.decode_limit},
+        )
+        decode_context = trace_context.child(
+            pipeline_job_id=decode_job_id,
+            acquisition_run_id=acquisition.run_id,
+            stage="decode",
+        )
+        _mark_pipeline_run(db_config, pipeline_run_id, status="running", current_stage="decode")
+        with transaction(db_config) as conn:
+            acquisition_repo = PostgresAcquisitionRepository(conn)
+            decode_repo = PostgresDecodeRepository(conn)
+            decode_service = DecodeService(
+                settings=settings,
+                repository=decode_repo,
+                trace_writer=trace_writer,
+                trace_context=decode_context,
+            )
+            decode = run_decode_stage(
+                candidate_repo=acquisition_repo,
+                decode_service=decode_service,
+                run_id=acquisition.run_id,
+                limit=args.decode_limit,
+                trace_writer=trace_writer,
+                trace_context=decode_context,
+            )
+            _mark_pipeline_job(
+                db_config,
+                decode_job_id,
+                status="partial" if decode.failed else "done",
+                metadata={"decode_total": decode.total, "decode_failed": decode.failed},
+            )
+            ingest_job_id = _start_pipeline_job(
+                db_config,
+                pipeline_run_id,
+                stage="ingest",
+                target_table="acquisition_runs",
+                target_id=acquisition.run_id,
+                metadata={"real_ingest": args.real_ingest},
+            )
+            ingest_context = decode_context.child(pipeline_job_id=ingest_job_id, stage="ingest")
+            _mark_pipeline_run(db_config, pipeline_run_id, status="running", current_stage="ingest")
+            ingest_records = [] if args.no_dry_ingest_record else _ingest_payloads(
+                decode_repo,
+                decode.outputs,
+                dry_run=not args.real_ingest,
+                env_file=args.env_file,
+                trace_writer=trace_writer,
+                trace_context=ingest_context,
+            )
+            ingest_failed = sum(1 for record in ingest_records if record.get("status") == "failed")
+            _mark_pipeline_job(
+                db_config,
+                ingest_job_id,
+                status="partial" if ingest_failed else "done",
+                metadata={"ingest_record_count": len(ingest_records), "failed": ingest_failed},
+            )
+        final_status = "partial" if acquisition.failed or decode.failed else "done"
+        _mark_pipeline_run(
+            db_config,
+            pipeline_run_id,
+            status=final_status,
+            current_stage="ingest",
+            metadata={
+                "acquisition_run_id": str(acquisition.run_id),
+                "decode_total": decode.total,
+                "decode_failed": decode.failed,
+                "ingest_record_count": len(ingest_records),
+            },
+        )
+        trace_writer.event(
+            context=trace_context.child(acquisition_run_id=acquisition.run_id),
+            stage="ingest",
+            event_type="pipeline_finished",
+            status=final_status,
+            payload={"decode": _model_dump(decode), "ingest_record_count": len(ingest_records)},
+        )
+    except Exception as exc:
+        _mark_pipeline_run(
+            db_config,
+            pipeline_run_id,
+            status="failed",
+            error_message=str(exc),
         )
-        ingest_records = [] if args.no_dry_ingest_record else _ingest_payloads(
-            decode_repo,
-            decode.outputs,
-            dry_run=not args.real_ingest,
-            env_file=args.env_file,
+        trace_writer.event(
+            context=trace_context,
+            stage="ingest",
+            event_type="pipeline_failed",
+            status="failed",
+            severity="error",
+            error_message=str(exc),
         )
+        raise
 
     print(json.dumps(
         {
             "batch_id": str(batch_id),
+            "pipeline_run_id": str(pipeline_run_id) if pipeline_run_id else None,
             "run_id": str(acquisition.run_id),
             "platforms": list(platforms),
             "acquisition": _model_dump(acquisition),

tests/test_app_api.py

@@ -322,6 +322,33 @@ class FakePipelineRepo:
             current_stage="decode",
         )
 
+    def get_pipeline_run_by_acquisition_run(self, acquisition_run_id):
+        return PipelineRun(
+            id=uuid4(),
+            run_key=f"pipeline-for-{acquisition_run_id}",
+            status="running",
+            current_stage="decode",
+        )
+
+    def list_timeline(self, run_id):
+        return [
+            {
+                "id": str(uuid4()),
+                "pipeline_run_id": str(run_id),
+                "stage": "decode",
+                "event_type": "decode_item_finished",
+                "status": "done",
+            }
+        ]
+
+    def get_timeline_bundle(self, run_id):
+        return {
+            "events": self.list_timeline(run_id),
+            "jobs": [{"id": str(uuid4()), "stage": "decode", "status": "done"}],
+            "candidate_hits": [{"id": str(uuid4()), "platform": "xiaohongshu"}],
+            "llm_call_traces": [{"id": str(uuid4()), "stage": "decode", "status": "success"}],
+        }
+
 
 class FakeManualRepo:
     def __init__(self):
@@ -391,6 +418,7 @@ def test_manual_query_batch_api_creates_ready_batch_and_starts_pipeline(monkeypa
         data = response.json()
         assert data["status"] == "queued"
         assert data["batch_id"] == str(repo.batch_id)
+        assert "pipeline_run_id" in data
         assert data["run_id"] == str(repo.run_id)
         assert data["pid"] == 12345
         assert data["query_count"] == 2
@@ -534,11 +562,28 @@ def test_formal_app_does_not_expose_legacy_static_mounts():
     assert "/api/creation-search/query" not in paths
 
 
-def test_pipeline_route_returns_501_instead_of_legacy_data_when_not_wired():
+def test_pipeline_routes_return_trace_run_and_timeline():
+    pipeline_repo = FakePipelineRepo()
+    app.dependency_overrides[get_pipeline_repository] = lambda: pipeline_repo
     client = TestClient(app)
     run_id = uuid4()
+    acquisition_run_id = uuid4()
+
+    try:
+        run = client.get(f"/api/pipeline/runs/{run_id}").json()["run"]
+        assert run["id"] == str(run_id)
+        assert run["current_stage"] == "decode"
+
+        timeline = client.get(f"/api/pipeline/runs/{run_id}/timeline").json()
+        assert timeline["events"][0]["event_type"] == "decode_item_finished"
+        assert timeline["jobs"][0]["stage"] == "decode"
+        assert timeline["candidate_hits"][0]["platform"] == "xiaohongshu"
+        assert timeline["llm_call_traces"][0]["status"] == "success"
 
-    assert client.get(f"/api/pipeline/runs/{run_id}").status_code == 501
+        by_acq = client.get(f"/api/pipeline/runs/by-acquisition/{acquisition_run_id}").json()["run"]
+        assert by_acq["run_key"] == f"pipeline-for-{acquisition_run_id}"
+    finally:
+        app.dependency_overrides.clear()
 
 
 def test_formal_decode_payload_pipeline_routes_have_override_success_paths():

tests/test_db_migration_contract.py

@@ -34,7 +34,13 @@ def test_creation_knowledge_migration_declares_formal_business_tables():
         "scope_results",
         "payload_drafts",
         "ingest_records",
-        "contract_snapshots",
+        "pipeline_runs",
+        "pipeline_jobs",
+        "pipeline_run_events",
+        "candidate_item_hits",
+        "llm_call_traces",
+        "contract_artifacts",
+        "run_contract_artifacts",
     ]
 
     assert "CREATE SCHEMA IF NOT EXISTS creation_knowledge" in sql

tests/test_decode_contracts.py

@@ -8,7 +8,7 @@ from decode_content.contracts import (
     SCOPE_TYPES,
     WHAT_KINDS,
     compute_contract_hash,
-    iter_contract_snapshots,
+    iter_contract_artifacts,
     load_decode_contracts,
 )
 
@@ -35,8 +35,8 @@ def test_contract_constants_record_formal_business_terms():
     assert DIM_ATTRIBUTE_BY_TYPE == {"how": "how", "what": "what", "why": "why"}
 
 
-def test_contract_snapshots_can_feed_repository():
-    snapshots = iter_contract_snapshots()
+def test_contract_artifacts_can_feed_repository():
+    snapshots = iter_contract_artifacts()
 
     assert snapshots[0].contract_name == "创作知识提取-skill"
     assert snapshots[0].contract_type == "skill"

tests/test_decode_service.py

@@ -45,6 +45,7 @@ class FakeRepo:
         return payload
 
     def save_contract_snapshot(self, **kwargs):
+        kwargs.pop("pipeline_run_id", None)
         snapshot = ContractSnapshot(id=uuid4(), **kwargs)
         self.snapshots.append(snapshot)
         return snapshot

tests/test_decode_stage_full_chain.py

@@ -88,6 +88,7 @@ class DecodeRepo:
         return payload
 
     def save_contract_snapshot(self, **kwargs):
+        kwargs.pop("pipeline_run_id", None)
         snapshot = ContractSnapshot(id=uuid4(), **kwargs)
         self.snapshots.append(snapshot)
         return snapshot

tests/test_ingest_payloads_script.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+from decode_content.models import IngestRecord, PayloadDraft
+from scripts.ingest_payloads import ingest_existing_payloads, list_payload_drafts_for_run
+
+
+def _payload():
+    return {
+        "source": {"id": "xhs_1", "source_type": "post", "title": "帖子"},
+        "title": "街拍姿势",
+        "content": "侧身站立可以拉长身体线条。",
+        "dim_attributes": ["how"],
+        "dim_creations": ["创作"],
+        "scopes": [{"scope_type": "effect", "value": "显高"}],
+        "custom_ext": [{"key": "业务阶段", "type": "str", "value": "灵感"}],
+    }
+
+
+class FakeRepo:
+    def __init__(self):
+        self.queries = []
+        self.marked = []
+        self.records = []
+
+    def _all(self, sql, params):
+        self.queries.append((sql, params))
+        return [
+            {
+                "id": uuid4(),
+                "particle_id": None,
+                "item_id": uuid4(),
+                "payload": _payload(),
+                "review_status": "pending",
+                "ingest_ready": False,
+                "status": "draft",
+                "error_message": None,
+                "created_at": None,
+                "updated_at": None,
+            }
+        ]
+
+    def mark_payload_draft_ingested(self, payload_draft_id):
+        self.marked.append(payload_draft_id)
+        return PayloadDraft(
+            id=payload_draft_id,
+            payload=_payload(),
+            review_status="approved",
+            ingest_ready=True,
+            status="ingested",
+        )
+
+    def save_ingest_record(self, **kwargs):
+        record = IngestRecord(id=uuid4(), **kwargs)
+        self.records.append(record)
+        return record
+
+
+def test_list_payload_drafts_for_run_skips_real_ingested_by_default():
+    repo = FakeRepo()
+    run_id = uuid4()
+
+    drafts = list_payload_drafts_for_run(repo, run_id=run_id)
+
+    sql, params = repo.queries[0]
+    assert drafts and isinstance(drafts[0], PayloadDraft)
+    assert params == (run_id,)
+    assert "target_system = 'knowledge-ingest-api'" in sql
+    assert "ir.status = 'ingested'" in sql
+
+
+def test_list_payload_drafts_for_run_can_include_real_ingested():
+    repo = FakeRepo()
+
+    list_payload_drafts_for_run(repo, run_id=uuid4(), include_real_ingested=True)
+
+    sql, _params = repo.queries[0]
+    assert "target_system = 'knowledge-ingest-api'" not in sql
+
+
+def test_ingest_existing_payloads_dry_run_records_without_http():
+    repo = FakeRepo()
+    draft = PayloadDraft(id=uuid4(), payload=_payload(), review_status="pending", ingest_ready=False, status="draft")
+
+    records = ingest_existing_payloads(repo, [draft], dry_run=True, env_file=".env")
+
+    assert repo.marked == [draft.id]
+    assert records[0]["target_system"] == "dry-run"
+    assert records[0]["status"] == "ingested"

tests/test_pipeline_tracing.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+from pathlib import Path
+from contextlib import nullcontext
+from uuid import uuid4
+
+from acquisition.domain import CandidateItem, MediaAsset
+from core.models import Post
+from decode_content.models import GateResult, ReadResult
+from decode_content.service import DecodeWorkflowOutput
+from pipeline.decode_runner import run_decode_stage
+from pipeline.tracing import TraceConfig, TraceContext, sanitize_payload
+from scripts import backfill_pipeline_traces
+
+
+def test_formal_schema_declares_append_only_trace_tables():
+    text = Path("db/migrations/001_creation_knowledge_schema.sql").read_text(encoding="utf-8")
+
+    for table in (
+        "pipeline_runs",
+        "pipeline_jobs",
+        "pipeline_run_events",
+        "candidate_item_hits",
+        "llm_call_traces",
+    ):
+        assert f"CREATE TABLE IF NOT EXISTS creation_knowledge.{table}" in text
+    assert "ALTER TABLE creation_knowledge.candidate_items" not in text
+    assert not Path("db/migrations/005_pipeline_trace_schema.sql").exists()
+
+
+def test_formal_schema_uses_deduped_contract_artifacts():
+    text = Path("db/migrations/001_creation_knowledge_schema.sql").read_text(encoding="utf-8")
+
+    assert "CREATE TABLE IF NOT EXISTS creation_knowledge.contract_artifacts" in text
+    assert "CREATE TABLE IF NOT EXISTS creation_knowledge.run_contract_artifacts" in text
+    assert "UNIQUE (contract_name, contract_type, source_path, content_hash)" in text
+    assert "CREATE TABLE IF NOT EXISTS creation_knowledge.contract_snapshots" not in text
+    assert not Path("db/migrations/005_pipeline_trace_schema.sql").exists()
+
+
+def test_postgres_decode_repository_writes_deduped_contract_artifacts():
+    text = Path("decode_content/repositories/postgres.py").read_text(encoding="utf-8")
+
+    assert "INSERT INTO contract_artifacts" in text
+    assert "INSERT INTO run_contract_artifacts" in text
+    assert "ON CONFLICT (contract_name, contract_type, source_path, content_hash)" in text
+    assert "INSERT INTO contract_snapshots" not in text
+
+
+def test_sanitize_payload_omits_base64_data_urls_but_keeps_metadata():
+    data_url = "data:video/mp4;base64," + ("a" * 120)
+
+    payload = sanitize_payload(
+        {"video": data_url, "url": "https://cdn.test/video.mp4"},
+        max_chars=50,
+    )
+
+    assert payload["url"] == "https://cdn.test/video.mp4"
+    assert payload["video"]["omitted"] == "base64_data_url"
+    assert payload["video"]["chars"] == len(data_url)
+    assert payload["video"]["sha256"]
+
+
+def test_trace_config_invalid_integer_falls_back_to_default(monkeypatch):
+    monkeypatch.setenv("CK_TRACE_MAX_PAYLOAD_CHARS", "not-a-number")
+
+    config = TraceConfig.from_env(".env.missing")
+
+    assert config.max_payload_chars == 50_000
+
+
+class CollectingTraceWriter:
+    def __init__(self):
+        self.events = []
+        self.hits = []
+        self.calls = []
+
+    def event(self, *, context, stage, event_type, **kwargs):
+        self.events.append({"context": context, "stage": stage, "event_type": event_type, **kwargs})
+
+    def candidate_hit(self, *, context, **kwargs):
+        self.hits.append({"context": context, **kwargs})
+
+    def llm_call(self, *, context, stage, substage=None, **kwargs):
+        self.calls.append({"context": context, "stage": stage, "substage": substage, **kwargs})
+
+
+class CandidateRepo:
+    def __init__(self):
+        self.item_id = uuid4()
+        self.job_id = uuid4()
+        self.query_id = uuid4()
+
+    def list_creation_candidate_items(self, *, run_id=None, limit=100):
+        return [
+            CandidateItem(
+                id=self.item_id,
+                job_id=self.job_id,
+                query_id=self.query_id,
+                platform="xiaohongshu",
+                platform_item_id="x1",
+                unique_key="xhs:x1",
+                canonical_url="https://xhs.test/1",
+                title="拍照姿势",
+                status="candidate",
+            )
+        ]
+
+    def list_media_assets_for_item(self, item_id):
+        return [
+            MediaAsset(
+                id=uuid4(),
+                item_id=item_id,
+                media_type="image",
+                cdn_url="https://cdn.test/1.jpg",
+                position=1,
+                status="done",
+            )
+        ]
+
+
+class DecodeService:
+    def decode_post(self, *, item_id, post: Post):
+        assert post.image_urls == ["https://cdn.test/1.jpg"]
+        return DecodeWorkflowOutput(
+            item_id=item_id,
+            read_result=ReadResult(text="ok"),
+            gate_result=GateResult(passed=True),
+            status="decoded",
+        )
+
+
+def test_decode_stage_emits_trace_events_without_changing_result():
+    writer = CollectingTraceWriter()
+    repo = CandidateRepo()
+    pipeline_run_id = uuid4()
+
+    result = run_decode_stage(
+        candidate_repo=repo,
+        decode_service=DecodeService(),
+        trace_writer=writer,
+        trace_context=TraceContext(pipeline_run_id=pipeline_run_id, stage="decode"),
+    )
+
+    assert result.decoded == 1
+    event_types = [event["event_type"] for event in writer.events]
+    assert event_types[0] == "decode_stage_started"
+    assert "decode_item_started" in event_types
+    assert "decode_item_finished" in event_types
+    assert event_types[-1] == "decode_stage_finished"
+    assert writer.events[1]["context"].pipeline_run_id == pipeline_run_id
+
+
+def test_backfill_pipeline_traces_dry_run_does_not_write(monkeypatch):
+    acquisition_run_id = uuid4()
+
+    monkeypatch.setattr(backfill_pipeline_traces.CreationDbConfig, "from_env", lambda _env_file: object())
+    monkeypatch.setattr(backfill_pipeline_traces, "transaction", lambda _config: nullcontext(object()))
+    monkeypatch.setattr(
+        backfill_pipeline_traces,
+        "_acquisition_runs",
+        lambda _conn, limit=None: [
+            {
+                "id": acquisition_run_id,
+                "run_key": "manual-api:test",
+                "status": "done",
+                "job_count": 3,
+                "candidate_count": 12,
+            }
+        ],
+    )
+
+    result = backfill_pipeline_traces.backfill(env_file=".env", apply=False)
+
+    assert result["dry_run"] is True
+    assert result["acquisition_run_count"] == 1
+    assert result["candidate_hit_count"] == 12
+    assert result["runs"][0]["acquisition_run_id"] == str(acquisition_run_id)

开发文档/正式版施工文档.md

@@ -214,7 +214,7 @@ db/migrations/001_creation_knowledge_schema.sql
 | `scope_results` | 作用域定位：实质、形式、感受、作用、意图 |
 | `payload_drafts` | 待审核或待入库 payload |
 | `ingest_records` | 入正式知识库记录 |
-| `contract_snapshots` | skill、prompt、schema 版本快照 |
+| `contract_artifacts` / `run_contract_artifacts` | skill、prompt、schema 合同快照与 run 级引用 |
 
 字段原则：
 
@@ -584,7 +584,7 @@ decode_content/contracts.py
 - 加载 `phase3-assemble.md`。
 - 加载 schema、taxonomy、prompt。
 - 计算合同版本 hash。
-- 写入 `contract_snapshots`。
+- 按内容 hash 写入 `contract_artifacts`，并在 `run_contract_artifacts` 里记录本次 run 使用的合同版本。
 
 原则：