Fix contract artifact timestamp mapping

decode_content/models.py

@@ -141,6 +141,7 @@ class ContractSnapshot(DecodeModel):
     source_path: str | None = None
     snapshot: JsonDict = Field(default_factory=dict)
     created_at: datetime | None = None
+    updated_at: datetime | None = None
 
 
 Knowledge = KnowledgeParticle

decode_content/repositories/postgres.py

@@ -371,7 +371,8 @@ class PostgresDecodeRepository:
                 source_path = COALESCE(EXCLUDED.source_path, contract_artifacts.source_path),
                 snapshot = EXCLUDED.snapshot,
                 updated_at = now()
-            RETURNING *
+            RETURNING id, contract_name, contract_type, version_label,
+                content_hash, source_path, snapshot, created_at, updated_at
             """,
             (
                 contract_name,

tests/test_decode_contracts.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+from datetime import datetime, timezone
+from uuid import uuid4
+
 from decode_content.contracts import (
     BUSINESS_STAGES,
     CREATION_STAGES,
@@ -11,6 +14,7 @@ from decode_content.contracts import (
     iter_contract_artifacts,
     load_decode_contracts,
 )
+from decode_content.models import ContractSnapshot
 
 
 def test_load_decode_contracts_covers_skill_prompts_and_cache():
@@ -45,3 +49,23 @@ def test_contract_artifacts_can_feed_repository():
     assert any(s.contract_type == "prompt_phase" for s in snapshots)
     assert any(s.contract_type == "schema" for s in snapshots)
     assert any(s.contract_type == "prompt" for s in snapshots)
+
+
+def test_contract_snapshot_accepts_contract_artifact_timestamps():
+    now = datetime.now(timezone.utc)
+
+    snapshot = ContractSnapshot.model_validate(
+        {
+            "id": uuid4(),
+            "contract_name": "创作知识提取-skill",
+            "contract_type": "skill",
+            "version_label": "test",
+            "content_hash": "a" * 64,
+            "source_path": "创作知识提取-skill",
+            "snapshot": {},
+            "created_at": now,
+            "updated_at": now,
+        }
+    )
+
+    assert snapshot.updated_at == now

tests/test_pipeline_tracing.py

@@ -1,18 +1,26 @@
 from __future__ import annotations
 
-from pathlib import Path
 from contextlib import nullcontext
+from datetime import datetime, timezone
+from pathlib import Path
 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.repositories.postgres import PostgresDecodeRepository
 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 _method_source(text: str, method_name: str) -> str:
+    start = text.index(f"    def {method_name}")
+    next_method = text.find("\n    def ", start + 1)
+    return text[start:] if next_method == -1 else text[start:next_method]
+
+
 def test_formal_schema_declares_append_only_trace_tables():
     text = Path("db/migrations/001_creation_knowledge_schema.sql").read_text(encoding="utf-8")
 
@@ -40,13 +48,61 @@ def test_formal_schema_uses_deduped_contract_artifacts():
 
 def test_postgres_decode_repository_writes_deduped_contract_artifacts():
     text = Path("decode_content/repositories/postgres.py").read_text(encoding="utf-8")
+    method = _method_source(text, "save_contract_snapshot")
 
-    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_artifacts" in method
+    assert "INSERT INTO run_contract_artifacts" in method
+    assert "ON CONFLICT (contract_name, contract_type, source_path, content_hash)" in method
+    assert "RETURNING id, contract_name, contract_type, version_label" in method
+    assert "content_hash, source_path, snapshot, created_at, updated_at" in method
     assert "INSERT INTO contract_snapshots" not in text
 
 
+class RecordingDecodeRepo(PostgresDecodeRepository):
+    def __init__(self):
+        super().__init__(conn=None)
+        self.one_calls = []
+        self.one_results = []
+
+    def _one(self, sql, params):
+        self.one_calls.append((sql, params))
+        return self.one_results.pop(0)
+
+
+def test_save_contract_snapshot_accepts_updated_at_from_contract_artifacts():
+    now = datetime.now(timezone.utc)
+    artifact_id = uuid4()
+    repo = RecordingDecodeRepo()
+    repo.one_results.append(
+        {
+            "id": artifact_id,
+            "contract_name": "创作知识提取-skill",
+            "contract_type": "skill",
+            "version_label": None,
+            "content_hash": "a" * 64,
+            "source_path": "创作知识提取-skill",
+            "snapshot": {},
+            "created_at": now,
+            "updated_at": now,
+        }
+    )
+
+    snapshot = repo.save_contract_snapshot(
+        contract_name="创作知识提取-skill",
+        contract_type="skill",
+        content_hash="a" * 64,
+        source_path="创作知识提取-skill",
+        snapshot={},
+    )
+
+    sql, _params = repo.one_calls[0]
+    assert "INSERT INTO contract_artifacts" in sql
+    assert "RETURNING *" not in sql
+    assert "updated_at" in sql
+    assert snapshot.id == artifact_id
+    assert snapshot.updated_at == now
+
+
 def test_sanitize_payload_omits_base64_data_urls_but_keeps_metadata():
     data_url = "data:video/mp4;base64," + ("a" * 120)