feat(pipeline): run real creation singleton to payload

acquisition/repositories/base.py

@@ -90,6 +90,16 @@ class AcquisitionRepository(Protocol):
     ) -> AcquisitionJob:
         ...
 
+    def update_acquisition_run(
+        self,
+        run_id: UUID,
+        *,
+        status: str,
+        error_message: str | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> AcquisitionRun:
+        ...
+
     def upsert_candidate_item(
         self,
         *,

acquisition/repositories/postgres.py

@@ -149,15 +149,16 @@ class PostgresAcquisitionRepository:
     ) -> AcquisitionRun:
         row = self._one(
             """
-            INSERT INTO acquisition_runs(batch_id, run_key, status, note, metadata)
-            VALUES (%s, %s, %s, %s, %s)
+            INSERT INTO acquisition_runs(batch_id, run_key, status, note, metadata, started_at)
+            VALUES (%s, %s, %s, %s, %s, CASE WHEN %s = 'running' THEN now() ELSE NULL END)
             ON CONFLICT (run_key) DO UPDATE SET
                 status = EXCLUDED.status,
                 note = EXCLUDED.note,
-                metadata = EXCLUDED.metadata
+                metadata = EXCLUDED.metadata,
+                started_at = COALESCE(acquisition_runs.started_at, EXCLUDED.started_at)
             RETURNING *
             """,
-            (batch_id, run_key, status, note, Json(metadata or {})),
+            (batch_id, run_key, status, note, Json(metadata or {}), status),
         )
         return AcquisitionRun.model_validate(row)
 
@@ -227,6 +228,35 @@ class PostgresAcquisitionRepository:
         )
         return AcquisitionJob.model_validate(row)
 
+    def update_acquisition_run(
+        self,
+        run_id: UUID,
+        *,
+        status: str,
+        error_message: str | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> AcquisitionRun:
+        row = self._one(
+            """
+            UPDATE acquisition_runs 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') THEN now() ELSE finished_at END
+            WHERE id = %s
+            RETURNING *
+            """,
+            (
+                status,
+                error_message,
+                metadata is not None,
+                Json(metadata or {}),
+                status,
+                run_id,
+            ),
+        )
+        return AcquisitionRun.model_validate(row)
+
     def upsert_candidate_item(
         self,
         *,

acquisition/runner.py

@@ -290,6 +290,27 @@ def run_batch(
                     },
                 )
 
+    run_status = (
+        "done"
+        if failed == 0 and partial == 0
+        else "partial"
+        if done > 0 or partial > 0
+        else "failed"
+    )
+    update_run = getattr(repo, "update_acquisition_run", None)
+    if update_run:
+        update_run(
+            run.id,
+            status=run_status,
+            metadata={
+                "total_jobs": total_jobs,
+                "done": done,
+                "partial": partial,
+                "failed": failed,
+                "skipped": skipped,
+            },
+        )
+
     return RunBatchResult(
         run_id=run.id,
         total_jobs=total_jobs,

app/dependencies.py

@@ -9,6 +9,7 @@ from fastapi import HTTPException
 from acquisition.repositories.postgres import PostgresAcquisitionRepository
 from core.config import CreationDbConfig
 from core.db_session import transaction
+from decode_content.repositories.postgres import PostgresDecodeRepository
 
 
 def _env_file() -> str:
@@ -24,8 +25,9 @@ def get_acquisition_repository() -> Iterator[PostgresAcquisitionRepository]:
         yield PostgresAcquisitionRepository(conn)
 
 
-def get_decode_repository():
-    raise HTTPException(status_code=501, detail="decode repository is wired in pipeline Step 7")
+def get_decode_repository() -> Iterator[PostgresDecodeRepository]:
+    with transaction(get_creation_db_config()) as conn:
+        yield PostgresDecodeRepository(conn)
 
 
 def get_pipeline_repository():

app/frontend/dist/index.html

@@ -4,8 +4,8 @@
     <meta charset="UTF-8" />
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <title>创作知识 · Query 正交 Demo</title>
-    <script type="module" crossorigin src="/app/assets/index-DEWR-s1T.js"></script>
-    <link rel="stylesheet" crossorigin href="/app/assets/index-CkGW0KOY.css">
+    <script type="module" crossorigin src="/app/assets/index-CFx5IMTf.js"></script>
+    <link rel="stylesheet" crossorigin href="/app/assets/index-B-5DKb0w.css">
   </head>
   <body>
     <div id="root"></div>

app/frontend/src/App.jsx

@@ -1,6 +1,7 @@
 import React, { useEffect, useState } from 'react'
 import CreationDemo from './pages/CreationDemo.jsx'
 import CreationQueryDetail from './pages/CreationQueryDetail.jsx'
+import DecodeItemDetail from './pages/DecodeItemDetail.jsx'
 
 function parseRoute() {
   const hash = window.location.hash || '#/'
@@ -8,6 +9,10 @@ function parseRoute() {
     const [, runId, queryId] = hash.match(/^#\/query\/([^/]+)\/([^/]+)/) || []
     return { name: 'query', runId: decodeURIComponent(runId || ''), queryId: decodeURIComponent(queryId || '') }
   }
+  if (hash.startsWith('#/decode-item/')) {
+    const [, itemId] = hash.match(/^#\/decode-item\/([^/]+)/) || []
+    return { name: 'decodeItem', itemId: decodeURIComponent(itemId || '') }
+  }
   return { name: 'demo' }
 }
 
@@ -22,6 +27,8 @@ export default function App() {
     <div className="wrap">
       {route.name === 'query' ? (
         <CreationQueryDetail runId={route.runId} queryId={route.queryId} />
+      ) : route.name === 'decodeItem' ? (
+        <DecodeItemDetail itemId={route.itemId} />
       ) : (
         <CreationDemo />
       )}

app/frontend/src/pages/CreationQueryDetail.jsx

@@ -83,6 +83,9 @@ function ResultCard({ item, tone, onImageClick }) {
       </div>
       {item.author_name && <div className="meta">{item.author_name}</div>}
       {item.classification?.reason && <div className="meta">{item.classification.reason}</div>}
+      {item.classification?.is_creation_knowledge === true && (
+        <a className="src detail-link" href={`#/decode-item/${encodeURIComponent(item.id)}`}>查看知识颗</a>
+      )}
       {item.raw_summary && (
         <details className="fold">
           <summary>正文摘要</summary>

app/frontend/src/pages/DecodeItemDetail.jsx

@@ -0,0 +1,257 @@
+import React, { useEffect, useMemo, useState } from 'react'
+
+function mediaUrl(asset) {
+  return asset?.cdn_url || asset?.oss_url || asset?.source_url || ''
+}
+
+function compactJson(value) {
+  return JSON.stringify(value || {}, null, 2)
+}
+
+function particleTone(type) {
+  return {
+    what: 'what',
+    how: 'how',
+    why: 'why',
+  }[type] || 'what'
+}
+
+function typeLabel(type) {
+  return {
+    what: 'What I know',
+    how: 'How I do',
+    why: 'Why it works',
+  }[type] || type || 'Knowledge'
+}
+
+function payloadTitle(payload) {
+  return payload?.title || payload?.source?.title || '未命名 payload'
+}
+
+function PayloadCard({ draft, ingest }) {
+  const payload = draft.payload || {}
+  return (
+    <article className="payload-card">
+      <div className="payload-head">
+        <b>{payloadTitle(payload)}</b>
+        <span className={`pill ${draft.status}`}>{draft.status}</span>
+      </div>
+      <div className="payload-meta">
+        <span>{(payload.dim_attributes || []).join(' / ') || '未标属性'}</span>
+        <span>{draft.ingest_ready ? '可入库' : '待审核'}</span>
+        {ingest && <span>{ingest.target_system} · {ingest.status}</span>}
+      </div>
+      {payload.content && <pre className="payload-content">{payload.content}</pre>}
+      {(payload.scopes || []).length > 0 && (
+        <div className="scope-row">
+          {payload.scopes.map((scope, idx) => (
+            <span className="scope-chip" key={`${scope.scope_type}-${scope.value}-${idx}`}>
+              {scope.scope_type}: {scope.value}
+            </span>
+          ))}
+        </div>
+      )}
+    </article>
+  )
+}
+
+function KnowledgeCard({ particle, scopes }) {
+  const content = particle.content || {}
+  const steps = content.steps || []
+  const body = content.body || []
+  return (
+    <article className={`knowledge-card ${particleTone(particle.particle_type)}`}>
+      <div className="knowledge-head">
+        <span>{typeLabel(particle.particle_type)}</span>
+        <b>{particle.title || '未命名知识颗'}</b>
+      </div>
+      <div className="knowledge-meta">
+        {particle.business_stage && <span>业务阶段：{particle.business_stage}</span>}
+        {particle.creation_stage && <span>创作阶段：{particle.creation_stage}</span>}
+        <span>{particle.status}</span>
+      </div>
+      {content.purpose && <p className="knowledge-purpose">{content.purpose}</p>}
+      {steps.length > 0 && (
+        <div className="step-list">
+          {steps.map((step, idx) => (
+            <div className="step-row" key={step.id || idx}>
+              <span className="step-no">{idx + 1}</span>
+              <div>
+                <div><b>输入</b>{step.input || '未写'}</div>
+                <div><b>指引</b>{step.directive || '未写'}</div>
+                <div><b>产出</b>{step.output || '未写'}</div>
+              </div>
+            </div>
+          ))}
+        </div>
+      )}
+      {body.length > 0 && (
+        <div className="what-body">
+          {body.map((row, idx) => (
+            <div className="what-row" key={`${row.item_name || 'row'}-${idx}`}>
+              <b>{row.item_name || `条目 ${idx + 1}`}</b>
+              <span>{row.item_desc || '暂无描述'}</span>
+            </div>
+          ))}
+        </div>
+      )}
+      {(content['阐述'] || content.desc) && (
+        <p className="knowledge-purpose">{content['阐述'] || content.desc}</p>
+      )}
+      {scopes.length > 0 && (
+        <div className="scope-row">
+          {scopes.map((scope) => (
+            <span className="scope-chip" key={scope.id}>
+              {scope.scope_type}: {scope.scope_value}
+            </span>
+          ))}
+        </div>
+      )}
+      <details className="fold knfold">
+        <summary>原始知识颗 JSON</summary>
+        <pre className="jsonbox">{compactJson(content)}</pre>
+      </details>
+    </article>
+  )
+}
+
+function SourcePanel({ item }) {
+  const media = item?.media_assets || []
+  const images = media.filter((asset) => ['image', 'cover', 'frame'].includes(asset.media_type)).map(mediaUrl).filter(Boolean)
+  const video = media.find((asset) => asset.media_type === 'video')
+  return (
+    <section className="source-panel">
+      <div>
+        <h1>{item?.title || 'Decode 知识详情'}</h1>
+        <div className="sub">
+          {item?.platform || 'unknown'} · {item?.author_name || '未知作者'} · {item?.content_type || '未知类型'}
+        </div>
+        {item?.raw_summary && <p className="source-summary">{item.raw_summary}</p>}
+        {item?.canonical_url && <a className="btn ghost" href={item.canonical_url} target="_blank" rel="noreferrer">打开原帖</a>}
+      </div>
+      <div className="source-media">
+        {video ? (
+          <video controls playsInline src={mediaUrl(video)} />
+        ) : images.length > 0 ? (
+          <div className="decode-thumbs">
+            {images.slice(0, 6).map((url, idx) => <img key={url} src={url} alt={`素材 ${idx + 1}`} loading="lazy" />)}
+          </div>
+        ) : (
+          <div className="media-empty">暂无媒体</div>
+        )}
+      </div>
+    </section>
+  )
+}
+
+export default function DecodeItemDetail({ itemId }) {
+  const [data, setData] = useState(null)
+  const [err, setErr] = useState('')
+
+  useEffect(() => {
+    setData(null)
+    setErr('')
+    if (!itemId) {
+      setErr('缺少 item_id')
+      return
+    }
+    fetch(`/api/decode/items/${encodeURIComponent(itemId)}/detail`)
+      .then((r) => (r.ok ? r.json() : Promise.reject(new Error('decode 详情读取失败'))))
+      .then(setData)
+      .catch((e) => setErr(e.message || '读取失败'))
+  }, [itemId])
+
+  const scopesByParticle = useMemo(() => {
+    const map = new Map()
+    for (const scope of data?.scope_results || []) {
+      const key = scope.particle_id || ''
+      map.set(key, [...(map.get(key) || []), scope])
+    }
+    return map
+  }, [data])
+
+  const ingestByDraft = useMemo(() => {
+    const map = new Map()
+    for (const record of data?.ingest_records || []) {
+      if (record.payload_draft_id) map.set(record.payload_draft_id, record)
+    }
+    return map
+  }, [data])
+
+  if (err) return <div className="empty">{err}</div>
+  if (!data) return <div className="empty">加载中...</div>
+
+  const readText = data.decode_result?.read_result?.text || ''
+  const particles = data.knowledge_particles || []
+  const drafts = data.payload_drafts || []
+
+  return (
+    <div>
+      <div className="detail-top">
+        <a className="back" href="#/">返回 Query 列表</a>
+        <span className="meta">item_id: {itemId}</span>
+      </div>
+      <SourcePanel item={data.item} />
+
+      <section className="decode-strip">
+        <div>
+          <span>Decode</span>
+          <b>{data.decode_result.status}</b>
+        </div>
+        <div>
+          <span>Knowledge</span>
+          <b>{particles.length}</b>
+        </div>
+        <div>
+          <span>Payload</span>
+          <b>{drafts.length}</b>
+        </div>
+        <div>
+          <span>Ingest</span>
+          <b>{data.ingest_records?.filter((row) => row.status === 'ingested').length || 0}</b>
+        </div>
+      </section>
+
+      <section className="decode-section">
+        <h2>单帖读懂</h2>
+        <div className="read-panel">{readText || '暂无读懂文本'}</div>
+        <div className="gate-panel">
+          <span className={`pill ${data.decode_result.gate_result?.passed ? 'decoded' : 'rejected'}`}>
+            {data.decode_result.gate_result?.passed ? '通过创作闸' : '未通过创作闸'}
+          </span>
+          <span>{data.decode_result.gate_result?.reason || '暂无判断理由'}</span>
+        </div>
+      </section>
+
+      <section className="decode-section">
+        <h2>知识颗</h2>
+        {particles.length === 0 ? (
+          <div className="row-empty">暂无 What/How/Why 知识颗</div>
+        ) : (
+          <div className="knowledge-grid">
+            {particles.map((particle) => (
+              <KnowledgeCard
+                key={particle.id}
+                particle={particle}
+                scopes={scopesByParticle.get(particle.id) || []}
+              />
+            ))}
+          </div>
+        )}
+      </section>
+
+      <section className="decode-section">
+        <h2>Payload Draft</h2>
+        {drafts.length === 0 ? (
+          <div className="row-empty">暂无 payload draft</div>
+        ) : (
+          <div className="payload-grid">
+            {drafts.map((draft) => (
+              <PayloadCard key={draft.id} draft={draft} ingest={ingestByDraft.get(draft.id)} />
+            ))}
+          </div>
+        )}
+      </section>
+    </div>
+  )
+}

app/frontend/src/styles.css

@@ -253,6 +253,7 @@ td.q { font-weight: 600; }
 .status.partial, .status.running { background: #fff6df; color: #9a6a00; }
 .status.failed { background: #fbeaea; color: var(--bad); }
 .src { display: inline-block; margin-top: 8px; font-size: 12px; }
+.detail-link { margin-left: 8px; font-weight: 700; }
 /* 平台区内：创作知识 / 非创作知识 小标题（各自起一行） */
 .subhd { display: flex; align-items: center; gap: 6px; font-size: 12px; font-weight: 700;
   color: var(--ink); margin: 14px 0 8px; }
@@ -265,8 +266,257 @@ td.q { font-weight: 600; }
 .lane.wx { border-left: 3px solid var(--ok); }
 .lane.xhs { border-left: 3px solid #ff2442; }
 
+/* decode 真实单例详情 */
+.source-panel {
+  display: grid;
+  grid-template-columns: minmax(0, 1.15fr) minmax(280px, .85fr);
+  gap: 18px;
+  align-items: start;
+  margin: 12px 0 18px;
+}
+.source-summary {
+  color: #3f3f46;
+  margin: 12px 0;
+  white-space: pre-wrap;
+  max-width: 72ch;
+}
+.source-media {
+  background: #fff;
+  border: 1px solid var(--line);
+  border-radius: 10px;
+  padding: 10px;
+  min-height: 180px;
+}
+.source-media video {
+  width: 100%;
+  max-height: 360px;
+  border-radius: 8px;
+  background: #000;
+}
+.decode-thumbs {
+  display: grid;
+  grid-template-columns: repeat(auto-fit, minmax(92px, 1fr));
+  gap: 8px;
+}
+.decode-thumbs img {
+  width: 100%;
+  aspect-ratio: 3 / 4;
+  object-fit: cover;
+  object-position: top;
+  border-radius: 8px;
+  border: 1px solid var(--line);
+  background: #f0f0f3;
+}
+.media-empty {
+  display: grid;
+  min-height: 160px;
+  place-items: center;
+  color: var(--muted);
+  background: #fafafb;
+  border-radius: 8px;
+}
+.decode-strip {
+  display: grid;
+  grid-template-columns: repeat(4, minmax(120px, 1fr));
+  gap: 10px;
+  margin-bottom: 20px;
+}
+.decode-strip > div {
+  background: #fff;
+  border: 1px solid var(--line);
+  border-radius: 10px;
+  padding: 10px 12px;
+}
+.decode-strip span {
+  display: block;
+  color: var(--muted);
+  font-size: 11.5px;
+  margin-bottom: 3px;
+}
+.decode-strip b { font-size: 18px; }
+.decode-section {
+  margin-top: 22px;
+}
+.decode-section h2 {
+  font-size: 16px;
+  margin: 0 0 10px;
+}
+.read-panel {
+  background: #fff;
+  border: 1px solid var(--line);
+  border-radius: 10px;
+  padding: 14px 16px;
+  color: #292932;
+  white-space: pre-wrap;
+  line-height: 1.72;
+  max-height: 360px;
+  overflow: auto;
+}
+.gate-panel {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 8px;
+  align-items: center;
+  margin-top: 10px;
+  color: var(--muted);
+}
+.pill {
+  display: inline-flex;
+  align-items: center;
+  border-radius: 999px;
+  padding: 2px 9px;
+  font-size: 11.5px;
+  font-weight: 700;
+  background: #f0f0f3;
+  color: #555;
+}
+.pill.decoded,
+.pill.ingested {
+  background: #e6f6ee;
+  color: var(--ok);
+}
+.pill.rejected,
+.pill.failed {
+  background: #fbeaea;
+  color: var(--bad);
+}
+.pill.draft,
+.pill.pending {
+  background: #fff6df;
+  color: #9a6a00;
+}
+.knowledge-grid,
+.payload-grid {
+  display: grid;
+  grid-template-columns: repeat(auto-fit, minmax(300px, 1fr));
+  gap: 12px;
+}
+.knowledge-card,
+.payload-card {
+  background: #fff;
+  border: 1px solid var(--line);
+  border-radius: 10px;
+  padding: 13px 14px;
+  min-width: 0;
+}
+.knowledge-card.what { border-color: #cfe8ff; }
+.knowledge-card.how { border-color: #d8eadf; }
+.knowledge-card.why { border-color: #eadfcf; }
+.knowledge-head,
+.payload-head {
+  display: flex;
+  gap: 8px;
+  justify-content: space-between;
+  align-items: start;
+  margin-bottom: 8px;
+}
+.knowledge-head span {
+  flex: 0 0 auto;
+  color: var(--accent);
+  background: var(--accent-soft);
+  border-radius: 7px;
+  padding: 2px 7px;
+  font-size: 11.5px;
+  font-weight: 700;
+}
+.knowledge-head b,
+.payload-head b {
+  font-size: 14px;
+  line-height: 1.45;
+}
+.knowledge-meta,
+.payload-meta {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 6px;
+  color: var(--muted);
+  font-size: 11.5px;
+  margin-bottom: 9px;
+}
+.knowledge-purpose {
+  margin: 8px 0 10px;
+  color: #33343c;
+  white-space: pre-wrap;
+}
+.step-list,
+.what-body {
+  display: grid;
+  gap: 8px;
+  margin-top: 8px;
+}
+.step-row {
+  display: grid;
+  grid-template-columns: 24px minmax(0, 1fr);
+  gap: 9px;
+  padding: 9px 10px;
+  background: #fafafb;
+  border: 1px solid #eeeef1;
+  border-radius: 8px;
+}
+.step-no {
+  width: 22px;
+  height: 22px;
+  border-radius: 50%;
+  display: inline-grid;
+  place-items: center;
+  color: #fff;
+  background: var(--accent);
+  font-size: 12px;
+  font-weight: 700;
+}
+.step-row b {
+  display: inline-block;
+  min-width: 34px;
+  color: #555;
+  margin-right: 8px;
+}
+.what-row {
+  display: grid;
+  gap: 4px;
+  padding: 9px 10px;
+  background: #fafafb;
+  border: 1px solid #eeeef1;
+  border-radius: 8px;
+}
+.scope-row {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 6px;
+  margin-top: 10px;
+}
+.scope-chip {
+  border-radius: 999px;
+  border: 1px solid var(--line);
+  background: #fafafb;
+  color: #555;
+  padding: 2px 8px;
+  font-size: 11.5px;
+}
+.jsonbox,
+.payload-content {
+  margin: 0;
+  padding: 10px;
+  border-radius: 8px;
+  background: #14151f;
+  color: #f1f1f5;
+  white-space: pre-wrap;
+  word-break: break-word;
+  font: 12px/1.65 ui-monospace, SFMono-Regular, Menlo, Consolas, monospace;
+  max-height: 280px;
+  overflow: auto;
+}
+.payload-content {
+  margin-top: 8px;
+  background: #fbfbfd;
+  color: #2f3038;
+  border: 1px solid var(--line);
+}
+
 @media (max-width: 720px) {
   .lanes { grid-template-columns: 1fr; }
+  .source-panel { grid-template-columns: 1fr; }
+  .decode-strip { grid-template-columns: repeat(2, minmax(0, 1fr)); }
+  .knowledge-grid, .payload-grid { grid-template-columns: 1fr; }
   .lightbox { padding: 18px 12px 48px; }
   .lightbox img { max-width: 96vw; max-height: 82vh; }
   .lbnav { bottom: 10px; top: auto; transform: none; }

app/frontend/vite.config.js

@@ -3,6 +3,8 @@ import react from '@vitejs/plugin-react'
 
 // 构建产物挂在 FastAPI 的 /app/ 下（base 必须对齐）。
 // dev 时把正式 API 请求代理到本地 uvicorn，省去跨域。
+const apiTarget = process.env.VITE_API_TARGET || 'http://127.0.0.1:8136'
+
 export default defineConfig({
   base: '/app/',
   plugins: [react()],
@@ -10,7 +12,7 @@ export default defineConfig({
   server: {
     port: 5180,
     proxy: {
-      '/api': 'http://127.0.0.1:8126',
+      '/api': apiTarget,
     },
   },
 })

app/routes/decode.py

@@ -6,8 +6,16 @@ from uuid import UUID
 
 from fastapi import APIRouter, Depends, HTTPException
 
-from app.dependencies import get_decode_repository
-from app.schemas import DecodeResultSchema
+from app.dependencies import get_acquisition_repository, get_decode_repository
+from app.schemas import (
+    CandidateItemSchema,
+    DecodeResultSchema,
+    IngestRecordSchema,
+    KnowledgeParticleSchema,
+    MediaAssetSchema,
+    PayloadDraftSchema,
+    ScopeResultSchema,
+)
 
 router = APIRouter(prefix="/api/decode", tags=["decode"])
 
@@ -23,6 +31,56 @@ def decode_result(
     return DecodeResultSchema.model_validate(getter(item_id))
 
 
+@router.get("/items/{item_id}/detail")
+def decode_item_detail(
+    item_id: UUID,
+    acquisition_repo: Any = Depends(get_acquisition_repository),
+    decode_repo: Any = Depends(get_decode_repository),
+) -> dict[str, Any]:
+    try:
+        item = acquisition_repo.get_candidate_item(item_id)
+        media = acquisition_repo.list_media_assets_for_item(item_id)
+        result = decode_repo.get_decode_result_for_item(item_id)
+    except Exception as exc:
+        raise HTTPException(status_code=404, detail=f"decode detail not found: {exc}") from exc
+
+    list_particles = getattr(decode_repo, "list_knowledge_particles", None)
+    list_scopes = getattr(decode_repo, "list_scope_results", None)
+    list_payloads = getattr(decode_repo, "list_payload_drafts", None)
+    list_ingests = getattr(decode_repo, "list_ingest_records", None)
+
+    raw_item = item.model_dump() if hasattr(item, "model_dump") else dict(item)
+    item_payload = CandidateItemSchema.model_validate(
+        {
+            **raw_item,
+            "media_assets": [
+                MediaAssetSchema.model_validate(row).model_dump(mode="json")
+                for row in media
+            ],
+        }
+    ).model_dump(mode="json")
+    return {
+        "item": item_payload,
+        "decode_result": DecodeResultSchema.model_validate(result).model_dump(mode="json"),
+        "knowledge_particles": [
+            KnowledgeParticleSchema.model_validate(row).model_dump(mode="json")
+            for row in (list_particles(item_id=item_id) if list_particles else [])
+        ],
+        "scope_results": [
+            ScopeResultSchema.model_validate(row).model_dump(mode="json")
+            for row in (list_scopes(item_id=item_id) if list_scopes else [])
+        ],
+        "payload_drafts": [
+            PayloadDraftSchema.model_validate(row).model_dump(mode="json")
+            for row in (list_payloads(item_id=item_id) if list_payloads else [])
+        ],
+        "ingest_records": [
+            IngestRecordSchema.model_validate(row).model_dump(mode="json")
+            for row in (list_ingests(item_id=item_id) if list_ingests else [])
+        ],
+    }
+
+
 @router.post("/items/{item_id}/jobs")
 def enqueue_decode_job(
     item_id: UUID,

decode_content/repositories/__init__.py

@@ -0,0 +1,5 @@
+"""Repository implementations for decode-content persistence."""
+
+from decode_content.repositories.postgres import PostgresDecodeRepository
+
+__all__ = ["PostgresDecodeRepository"]

decode_content/repositories/postgres.py

@@ -0,0 +1,366 @@
+"""PostgreSQL implementation for formal decode-content state."""
+from __future__ import annotations
+
+from typing import Any
+from uuid import UUID
+
+import psycopg2.extras
+
+from decode_content.models import (
+    ContractSnapshot,
+    DecodeJob,
+    DecodeResult,
+    IngestRecord,
+    KnowledgeParticle,
+    PayloadDraft,
+    ScopeResult,
+)
+
+Json = psycopg2.extras.Json
+psycopg2.extras.register_uuid()
+
+
+def _json(value: dict[str, Any] | None) -> Json:
+    return Json(value or {})
+
+
+def _stage_value(value: Any) -> str | None:
+    if isinstance(value, list):
+        return " / ".join(str(item) for item in value if item)
+    if value:
+        return str(value)
+    return None
+
+
+class PostgresDecodeRepository:
+    """Repository backed by the formal PostgreSQL schema.
+
+    Like the acquisition repository, this class owns no transaction boundary;
+    callers pass an open connection and commit/rollback outside.
+    """
+
+    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_decode_job(
+        self,
+        *,
+        item_id: UUID,
+        status: str = "pending",
+        metadata: dict[str, Any] | None = None,
+    ) -> DecodeJob:
+        row = self._one(
+            """
+            INSERT INTO decode_jobs(item_id, status, metadata, started_at)
+            VALUES (%s, %s, %s, CASE WHEN %s = 'running' THEN now() ELSE NULL END)
+            RETURNING *
+            """,
+            (item_id, status, _json(metadata), status),
+        )
+        return DecodeJob.model_validate(row)
+
+    def save_decode_result(
+        self,
+        *,
+        item_id: UUID,
+        decode_job_id: UUID | None = None,
+        read_result: dict[str, Any] | None = None,
+        gate_result: dict[str, Any] | None = None,
+        framing_result: dict[str, Any] | None = None,
+        status: str = "draft",
+    ) -> DecodeResult:
+        row = self._one(
+            """
+            INSERT INTO decode_results(
+                decode_job_id, item_id, read_result, gate_result,
+                framing_result, status
+            )
+            VALUES (%s, %s, %s, %s, %s, %s)
+            RETURNING *
+            """,
+            (
+                decode_job_id,
+                item_id,
+                _json(read_result),
+                _json(gate_result),
+                _json(framing_result),
+                status,
+            ),
+        )
+        if decode_job_id:
+            self._one(
+                """
+                UPDATE decode_jobs
+                SET status = %s, finished_at = now(), error_message = NULL
+                WHERE id = %s
+                RETURNING *
+                """,
+                (status, decode_job_id),
+            )
+        return DecodeResult.model_validate(row)
+
+    def get_decode_result_for_item(self, item_id: UUID) -> DecodeResult:
+        row = self._one(
+            """
+            SELECT * FROM decode_results
+            WHERE item_id = %s
+            ORDER BY created_at DESC
+            LIMIT 1
+            """,
+            (item_id,),
+        )
+        return DecodeResult.model_validate(row)
+
+    def save_knowledge_particle(
+        self,
+        *,
+        item_id: UUID,
+        particle_type: str,
+        title: str,
+        decode_result_id: UUID | None = None,
+        parent_particle_id: UUID | None = None,
+        content: dict[str, Any] | None = None,
+        status: str = "draft",
+    ) -> KnowledgeParticle:
+        content = content or {}
+        creation_stage = _stage_value(content.get("创作阶段"))
+        if creation_stage is None:
+            for step in content.get("steps") or []:
+                creation_stage = _stage_value(step.get("创作阶段"))
+                if creation_stage:
+                    break
+        row = self._one(
+            """
+            INSERT INTO knowledge_particles(
+                decode_result_id, item_id, parent_particle_id, particle_type,
+                title, business_stage, creation_stage, content, status
+            )
+            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
+            RETURNING *
+            """,
+            (
+                decode_result_id,
+                item_id,
+                parent_particle_id,
+                particle_type,
+                title,
+                _stage_value(content.get("业务阶段")),
+                creation_stage,
+                _json(content),
+                status,
+            ),
+        )
+        return KnowledgeParticle.model_validate(row)
+
+    def list_knowledge_particles(self, item_id: UUID | None = None) -> list[KnowledgeParticle]:
+        if item_id is None:
+            rows = self._all(
+                "SELECT * FROM knowledge_particles ORDER BY created_at DESC",
+                (),
+            )
+        else:
+            rows = self._all(
+                """
+                SELECT * FROM knowledge_particles
+                WHERE item_id = %s
+                ORDER BY sort_order, created_at
+                """,
+                (item_id,),
+            )
+        return [KnowledgeParticle.model_validate(row) for row in rows]
+
+    def save_scope_result(
+        self,
+        *,
+        scope_type: str,
+        scope_value: str,
+        particle_id: UUID | None = None,
+        item_id: UUID | None = None,
+        evidence: dict[str, Any] | None = None,
+        status: str = "draft",
+    ) -> ScopeResult:
+        evidence = evidence or {}
+        row = self._one(
+            """
+            INSERT INTO scope_results(
+                particle_id, item_id, scope_type, scope_value, is_reused,
+                matched_scope_id, confidence, evidence, status
+            )
+            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
+            RETURNING *
+            """,
+            (
+                particle_id,
+                item_id,
+                scope_type,
+                scope_value,
+                evidence.get("is_reused"),
+                evidence.get("matched_scope_id"),
+                evidence.get("confidence"),
+                _json(evidence),
+                status,
+            ),
+        )
+        return ScopeResult.model_validate(row)
+
+    def list_scope_results(self, item_id: UUID | None = None) -> list[ScopeResult]:
+        if item_id is None:
+            rows = self._all("SELECT * FROM scope_results ORDER BY created_at DESC", ())
+        else:
+            rows = self._all(
+                """
+                SELECT * FROM scope_results
+                WHERE item_id = %s
+                ORDER BY created_at
+                """,
+                (item_id,),
+            )
+        return [ScopeResult.model_validate(row) for row in rows]
+
+    def save_payload_draft(
+        self,
+        *,
+        payload: dict[str, Any],
+        particle_id: UUID | None = None,
+        item_id: UUID | None = None,
+        review_status: str = "pending",
+        ingest_ready: bool = False,
+        status: str = "draft",
+    ) -> PayloadDraft:
+        row = self._one(
+            """
+            INSERT INTO payload_drafts(
+                particle_id, item_id, payload, review_status, ingest_ready, status
+            )
+            VALUES (%s, %s, %s, %s, %s, %s)
+            RETURNING *
+            """,
+            (particle_id, item_id, _json(payload), review_status, ingest_ready, status),
+        )
+        return PayloadDraft.model_validate(row)
+
+    def list_payload_drafts(self, item_id: UUID | None = None) -> list[PayloadDraft]:
+        if item_id is None:
+            rows = self._all(
+                "SELECT * FROM payload_drafts ORDER BY created_at DESC",
+                (),
+            )
+        else:
+            rows = self._all(
+                """
+                SELECT * FROM payload_drafts
+                WHERE item_id = %s
+                ORDER BY created_at
+                """,
+                (item_id,),
+            )
+        return [PayloadDraft.model_validate(row) for row in rows]
+
+    def mark_payload_draft_ingested(self, payload_draft_id: UUID) -> PayloadDraft:
+        row = self._one(
+            """
+            UPDATE payload_drafts
+            SET review_status = 'approved',
+                ingest_ready = true,
+                status = 'ingested',
+                error_message = NULL
+            WHERE id = %s
+            RETURNING *
+            """,
+            (payload_draft_id,),
+        )
+        return PayloadDraft.model_validate(row)
+
+    def save_ingest_record(
+        self,
+        *,
+        payload_draft_id: UUID | None = None,
+        target_system: str,
+        target_id: str | None = None,
+        status: str = "pending",
+        response_payload: dict[str, Any] | None = None,
+        error_message: str | None = None,
+    ) -> IngestRecord:
+        row = self._one(
+            """
+            INSERT INTO ingest_records(
+                payload_draft_id, target_system, target_id, status,
+                attempt_count, response_payload, error_message
+            )
+            VALUES (%s, %s, %s, %s, 1, %s, %s)
+            RETURNING *
+            """,
+            (
+                payload_draft_id,
+                target_system,
+                target_id,
+                status,
+                _json(response_payload),
+                error_message,
+            ),
+        )
+        return IngestRecord.model_validate(row)
+
+    def list_ingest_records(self, item_id: UUID | None = None) -> list[IngestRecord]:
+        if item_id is None:
+            rows = self._all("SELECT * FROM ingest_records ORDER BY created_at DESC", ())
+        else:
+            rows = self._all(
+                """
+                SELECT ir.* FROM ingest_records ir
+                JOIN payload_drafts pd ON pd.id = ir.payload_draft_id
+                WHERE pd.item_id = %s
+                ORDER BY ir.created_at
+                """,
+                (item_id,),
+            )
+        return [IngestRecord.model_validate(row) for row in rows]
+
+    def save_contract_snapshot(
+        self,
+        *,
+        contract_name: str,
+        contract_type: str,
+        version_label: str | None = None,
+        content_hash: str | None = None,
+        source_path: str | None = None,
+        snapshot: dict[str, Any] | None = None,
+    ) -> ContractSnapshot:
+        row = self._one(
+            """
+            INSERT INTO contract_snapshots(
+                contract_name, contract_type, version_label, content_hash,
+                source_path, snapshot
+            )
+            VALUES (%s, %s, %s, %s, %s, %s)
+            RETURNING *
+            """,
+            (
+                contract_name,
+                contract_type,
+                version_label,
+                content_hash,
+                source_path,
+                _json(snapshot),
+            ),
+        )
+        return ContractSnapshot.model_validate(row)

scripts/run_creation_singleton.py

@@ -0,0 +1,193 @@
+#!/usr/bin/env python3
+"""Run one real f1/f2 creation-knowledge pipeline sample to dry-run ingest."""
+from __future__ import annotations
+
+import argparse
+import json
+from dataclasses import asdict, is_dataclass
+from datetime import datetime
+from pathlib import Path
+import sys
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from acquisition.queries.builder import (  # noqa: E402
+    QueryBuildOptions,
+    TREES,
+    build_creation_query_batch,
+    persist_query_batch,
+)
+from acquisition.repositories.postgres import PostgresAcquisitionRepository  # noqa: E402
+from acquisition.runner import DEFAULT_PLATFORMS, run_batch  # noqa: E402
+from core.config import CreationDbConfig, Settings  # noqa: E402
+from core.db_session import transaction  # noqa: E402
+from decode_content.repositories.postgres import PostgresDecodeRepository  # noqa: E402
+from decode_content.service import DecodeService  # noqa: E402
+from pipeline.decode_runner import run_decode_stage  # noqa: E402
+
+
+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("--tree-path", type=Path, default=TREES)
+    parser.add_argument("--per", type=int, default=1, help="Queries per active family")
+    parser.add_argument("--batch-n", type=int, default=30)
+    parser.add_argument("--seed", type=int, default=7)
+    parser.add_argument(
+        "--dry-query-filter",
+        action="store_true",
+        help="Skip only the query-filter LLM; search/classify/decode remain real.",
+    )
+    parser.add_argument(
+        "--platform",
+        action="append",
+        choices=DEFAULT_PLATFORMS,
+        help="Platform to run. Repeat for multiple platforms. Default: xiaohongshu.",
+    )
+    parser.add_argument("--search-limit", type=int, default=1)
+    parser.add_argument("--display-limit", type=int, default=1)
+    parser.add_argument("--decode-limit", type=int, default=1)
+    parser.add_argument("--name", default="")
+    parser.add_argument("--frontend-base", default="http://127.0.0.1:5180/app/")
+    return parser.parse_args(argv)
+
+
+def _now_key() -> str:
+    return datetime.now().strftime("%Y%m%d-%H%M%S")
+
+
+def _model_dump(value: Any) -> dict[str, Any]:
+    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 _dry_ingest_payloads(
+    repo: PostgresDecodeRepository,
+    outputs: list[Any],
+) -> list[dict[str, Any]]:
+    records: list[dict[str, Any]] = []
+    for output in outputs:
+        for draft in output.payload_drafts:
+            if draft.id is None:
+                continue
+            repo.mark_payload_draft_ingested(draft.id)
+            record = repo.save_ingest_record(
+                payload_draft_id=draft.id,
+                target_system="dry-run",
+                target_id=str(draft.id),
+                status="ingested",
+                response_payload={
+                    "dry_run": True,
+                    "note": "payload generated by singleton pipeline; external ingest API not called",
+                    "payload": draft.payload,
+                },
+            )
+            records.append(_model_dump(record))
+    return records
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    settings = Settings.from_env(args.env_file)
+    db_config = CreationDbConfig.from_env(args.env_file)
+    platforms = tuple(args.platform or ("xiaohongshu",))
+    run_key_suffix = _now_key()
+    name = args.name or f"creation-singleton-{run_key_suffix}"
+
+    generated = build_creation_query_batch(
+        settings,
+        tree_path=args.tree_path,
+        options=QueryBuildOptions(
+            per=args.per,
+            batch_n=args.batch_n,
+            seed=args.seed,
+            dry=args.dry_query_filter,
+            active_family_keys=("f1", "f2"),
+        ),
+    )
+    query_count = sum(len(family.get("items") or []) for family in generated.get("families") or [])
+    kept_count = sum(
+        1
+        for family in generated.get("families") or []
+        for item in family.get("items") or []
+        if item.get("keep", True)
+    )
+
+    with transaction(db_config) as conn:
+        acquisition_repo = PostgresAcquisitionRepository(conn)
+        batch, persisted_count = persist_query_batch(
+            acquisition_repo,
+            generated,
+            name=name,
+            source_type="generated",
+            generation_method="creation_singleton_v1",
+            target_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=False,
+            skip_done=False,
+            run_key=f"singleton-acquisition:{batch.id}:{run_key_suffix}",
+        )
+
+    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,
+        )
+        ingest_records = _dry_ingest_payloads(decode_repo, decode.outputs)
+
+    decoded_items = [str(output.item_id) for output in decode.outputs]
+    first_item_id = decoded_items[0] if decoded_items else None
+    summary = {
+        "batch_id": str(batch.id),
+        "run_id": str(acquisition.run_id),
+        "active_family_keys": ["f1", "f2"],
+        "query_count": query_count,
+        "kept_count": kept_count,
+        "persisted_queries": persisted_count,
+        "platforms": list(platforms),
+        "acquisition": _model_dump(acquisition),
+        "decode": {
+            "total": decode.total,
+            "decoded": decode.decoded,
+            "skipped": decode.skipped,
+            "failed": decode.failed,
+            "decoded_items": decoded_items,
+            "payload_count": sum(len(output.payload_drafts) for output in decode.outputs),
+        },
+        "dry_run_ingest_count": len(ingest_records),
+        "detail_url": (
+            f"{args.frontend_base.rstrip('/')}/#/decode-item/{first_item_id}"
+            if first_item_id
+            else None
+        ),
+    }
+    print(json.dumps(summary, ensure_ascii=False, indent=2, default=str))
+    return 0 if first_item_id else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_app_api.py

@@ -12,7 +12,14 @@ from app.dependencies import (
     get_decode_repository,
     get_pipeline_repository,
 )
-from decode_content.models import DecodeJob, DecodeResult, PayloadDraft
+from decode_content.models import (
+    DecodeJob,
+    DecodeResult,
+    IngestRecord,
+    KnowledgeParticle,
+    PayloadDraft,
+    ScopeResult,
+)
 from pipeline.models import PipelineRun
 
 
@@ -122,6 +129,14 @@ class FakeAcquisitionRepo:
             ],
         }
 
+    def get_candidate_item(self, item_id):
+        assert item_id == self.item_id
+        return self.get_query_detail(run_id=self.run_id, query_id=self.query_id)["items"][0]
+
+    def list_media_assets_for_item(self, item_id):
+        assert item_id == self.item_id
+        return self.get_query_detail(run_id=self.run_id, query_id=self.query_id)["media_assets"]
+
 
 class FakeDecodeRepo:
     def __init__(self):
@@ -159,6 +174,39 @@ class FakeDecodeRepo:
             )
         ]
 
+    def list_knowledge_particles(self, item_id=None):
+        return [
+            KnowledgeParticle(
+                id=uuid4(),
+                item_id=item_id or uuid4(),
+                particle_type="what",
+                title="What I know",
+                content={"content": "一个可复用知识点"},
+                status="draft",
+            )
+        ]
+
+    def list_scope_results(self, item_id=None):
+        return [
+            ScopeResult(
+                id=uuid4(),
+                item_id=item_id or uuid4(),
+                scope_type="intent",
+                scope_value="吸引注意",
+                status="draft",
+            )
+        ]
+
+    def list_ingest_records(self, item_id=None):
+        return [
+            IngestRecord(
+                id=uuid4(),
+                payload_draft_id=self.payload_id,
+                target_system="dry-run",
+                status="ingested",
+            )
+        ]
+
 
 class FakePipelineRepo:
     def get_pipeline_run(self, run_id):
@@ -205,22 +253,20 @@ def test_formal_app_does_not_expose_legacy_static_mounts():
     assert "/api/creation-search/query" not in paths
 
 
-def test_unimplemented_formal_routes_return_501_instead_of_legacy_data():
+def test_pipeline_route_returns_501_instead_of_legacy_data_when_not_wired():
     client = TestClient(app)
-    item_id = uuid4()
     run_id = uuid4()
 
-    assert client.get(f"/api/decode/items/{item_id}/result").status_code == 501
-    assert client.post(f"/api/decode/items/{item_id}/jobs").status_code == 501
-    assert client.get("/api/payloads/drafts").status_code == 501
     assert client.get(f"/api/pipeline/runs/{run_id}").status_code == 501
 
 
 def test_formal_decode_payload_pipeline_routes_have_override_success_paths():
+    acquisition_repo = FakeAcquisitionRepo()
     decode_repo = FakeDecodeRepo()
     pipeline_repo = FakePipelineRepo()
-    item_id = uuid4()
+    item_id = acquisition_repo.item_id
     run_id = uuid4()
+    app.dependency_overrides[get_acquisition_repository] = lambda: acquisition_repo
     app.dependency_overrides[get_decode_repository] = lambda: decode_repo
     app.dependency_overrides[get_pipeline_repository] = lambda: pipeline_repo
     client = TestClient(app)
@@ -238,6 +284,11 @@ def test_formal_decode_payload_pipeline_routes_have_override_success_paths():
         assert drafts[0]["item_id"] == str(item_id)
         assert drafts[0]["payload"]["title"] == "可审核 payload"
 
+        detail = client.get(f"/api/decode/items/{item_id}/detail").json()
+        assert detail["item"]["id"] == str(item_id)
+        assert detail["knowledge_particles"][0]["title"] == "What I know"
+        assert detail["ingest_records"][0]["status"] == "ingested"
+
         run = client.get(f"/api/pipeline/runs/{run_id}").json()["run"]
         assert run["id"] == str(run_id)
         assert run["current_stage"] == "decode"