#AI 决策可追溯性：把 AI 做事过程变成可查的账

这不是“再多打一些 log”。我们要补的是一层业务账本：AI 看到了什么、准备帮员工做什么、系统规则为什么放行或拦住、最后有没有真的写进数据库。

推荐做法是保留现有主表作为当前状态，另外只给高影响 AI 判断补 append-only ledger。也就是 current-state 主表 + selective append-only AI/audit ledger。

当前还没有 production 用户，也没有生产历史需要保护。这个阶段应该一次把地基设计对；breaking change 可以接受，不需要为 test data 做复杂 backfill。

#1. 我们到底想解决什么

现在的问题不是“系统完全没有记录”。accepted task、call 分析结果、contact timeline 其实已经有一些记录。真正的问题是：当 AI 没有成功写出结果时，我们经常不知道中间发生了什么。

举个最具体的例子：一通电话里客户表达了取消风险，AI 可能建议创建一个 cancellation risk task。但最后页面上没有这个 task。今天我们很难快速判断：是 AI 根本没提出来？是 DNC 拦住了？是系统认为已经有重复 task？是引用的 call/source 解析失败？还是数据库写入时发生了 zero-row race？

这些情况在产品上看起来都像“没有 task”，但工程含义完全不同。它们对应不同的修复方向：prompt 要不要改、guard 是否过严、source linking 是否坏了、还是数据库写入路径有 bug。

所以我们要补的不是一堆散 log，而是一条可查询的决策流水。以后看到一个 AI 自动化结果，或者看到某个结果没有出现，我们都能沿着这条流水查回去。

这条流水要能回答四件事：

1. AI 当时准备帮业务做哪件事。 比如建议新建一个 cancellation risk task，或者建议关闭一个 follow-up task。
2. 这个建议是根据什么来的。 比如哪通 call、哪条 SMS、哪个 lead，或者哪次 contact analysis。
3. 系统规则怎么判。 比如允许、拒绝、跳过、需要人工复核；如果被拦住，要能看到原因。
4. 最后有没有真的写入。 成功写入、没尝试写、写入时发现目标已经变了、还是写入失败，这几个不能混在一起。

#2. 做完以后，大家会怎么用

这套东西做完以后，最直接的变化是：页面不再只展示“现在有什么”，还能解释“它为什么会在这里”。工程也不再靠翻 CloudWatch 猜测 AI 当时发生了什么。

产品和运营会看到的变化：

• Task 页面能看到它是由哪通 call、哪条 SMS、哪个 lead 触发的。
• 如果一个 task 被某通电话关闭或补强，页面也能显示这个关系。
• 就算 AI 提了建议但最后没生成 task，也能在 audit 视图里看到“提过什么、为什么没写入”。

工程和 PM 会得到的能力：

• 查 prompt 改版后，哪些 proposal 更容易被 reject 或 zero-row。
• 查某个 store 最近被 DNC、duplicate、low confidence 拦下的 AI 动作。
• 查同一通 call 在不同 prompt/model 下的判断是否变了。

对未来的 self-check agent 也很关键。一个 agent 如果要复核“这个 task 是否合理”，它不能只看 task 当前字段；它还要知道 task 是怎么来的、证据是什么、哪些建议曾经被拒绝过。

#3. 为什么现在就要做

现在还在 test-only 阶段，没有 production 用户，也没有需要长期保留的真实历史数据。这是改地基最便宜的时候。

如果等到 production 后再补，问题会变复杂：要兼容旧数据、做 backfill、保护线上读写、加 feature flag，还要解释历史上那些没有记录的 AI 行为。那时候不是不能做，但成本会高很多，而且容易留下长期技术债。

当前阶段建议直接按目标模型改。test data 可以 reset / truncate / rerun；不为了兼容测试历史保留错误抽象。等有 production 用户以后，再切换到 expand/contract、backfill、feature flag 这一套安全迁移方式。

#4. 我们选什么架构模式

这里有一个边界要想清楚：我们不是要把整个业务系统改造成 Event Sourcing。我们只是要让高影响 AI 判断有账可查。

推荐模式：contacts、tasks、calls 继续保当前状态；AI 关键判断额外写 append-only ledger。主表负责产品读，ledger 负责解释和审计。

为什么不只用 overwrite：

• overwrite 适合展示当前状态。
• 但它解释不了旧判断为什么被覆盖。
• 也解释不了 AI 想做但没做成的事情。

为什么不上 full Event Sourcing：

• full Event Sourcing 会把主表变成 projection。
• 读路径、replay、event schema versioning 都会变复杂。
• 我们现在缺的是 AI 审计账本，不是整个系统的事件源。

所以最稳的选择是 selective ledger：只把真正需要长期解释的 AI 决策写成 append-only 记录。低风险派生字段、缓存、展示分组，仍然可以 overwrite 或重新计算。

#5. 现有表怎么分工，避免重复造轮子

有些能力系统已经有了，不需要重新造。新表只补空白，不复制现有职责。

#6. 具体要做哪些表和字段

下面开始进入工程落地层。这里用表格是合适的，因为它回答的是“改哪张表、为什么改、优先级是什么”。

#7. 第一块最重要：task_decision_attempts

task_decision_attempts 不是另一张 tasks 表。它记录的是“一次 AI 对 task 的尝试”。尝试可能成功，也可能被规则拒绝、被跳过、写入失败。只要 AI 认真提出过，就应该留下记录。

一个典型记录长这样：AI 根据一通电话建议创建 cancellation risk task；系统发现已经有同类 open task，于是判定 duplicate；最后没有写新 task。这个事情不应该只出现在 CloudWatch，它应该能在 DB 里按 store、contact、call、promptVersion 查出来。

reject_reason 应该对齐 callytics-common/src/domain/types.ts 里的 RejectReason union，不要在文档或 migration 里手抄一份容易 stale 的字符串列表。

needs_review 是 Phase 2 reserved。shared type 已有，但 Phase 1 writer 当前不返回它。

#8. Call 和 Contact 怎么补 traceability

#8.1 Call analysis

calls 继续保最新结果，这个方向没问题。live code 已经把每次 call_analysis.completed 写进 contact_timeline，并用 callId + analysisRunId 做 per-run idempotency。所以 call verdict history 不是完全丢失。

真正缺的是一个专门给 call analysis 查 run 的 thin index，以及统一的 AI usage/cost ledger。也就是说：call_analysis_runs 只做 call run index；model、prompt、token、cost 去 ai_runs 查。

#8.2 Contact profile

对 contact 来说，最重要的是不要让“AI 推断”和“员工/CRM/POS 确认”长得一样。比如 lifecycleStage=member，它可能是 AI 从通话里推断出来的，也可能是员工确认的，也可能来自 POS。这几个来源的权威性不同。

• 高 source rank 覆盖低 source rank。
• 低 source rank 不能静默覆盖高 source rank。
• 同 source rank 时，新值覆盖旧值。
• *_trust_score 只是 source hierarchy 的数字编码，不是 AI confidence，也不是新的业务真理。

#9. Task 页面怎么形成联动

这套设计做完后，task 页面可以和 call、lead、message 形成真正联动。关键是分两类：已经写成 task 的，用 task_sources；没有写成 task 的，用 task_decision_attempts.resolved_sources。

• 已创建的 task： 页面展示它来自哪通 call、哪条 SMS、哪个 lead，以及后续哪些 source 支持或关闭了它。
• 没创建出来的 proposal： 在 call/contact audit 页面展示 AI 当时提过什么、为什么被拒绝或跳过。
• 从 source 反查： 打开一通 call 或一个 lead，也能看到它生成或影响了哪些 task。

第一版不需要为 rejected proposal 额外做高频 reverse lookup 表。resolved_sources 用 JSONB 足够支撑 audit 页。等它变成高频产品路径，再加 source-edge projection。

#10. 分阶段落地

#11. 明确不做什么

• 不做 full Event Sourcing。
• 不做泛泛的 ai_decision_log 大杂烩。
• 不为每个 contact field 单独建 history table。
• 不把 task 页面读路径改成 replay event。
• 不把未校准的 confidence 当成权威来源。
• 不把 CloudWatch log 当 durable ledger。

#12. 还需要拍板的问题

1. task_decision_attempts.proposal_json 保完整 normalized proposal，还是只保 subset + S3 pointer。
2. full prompt/output 放 S3、Langfuse/Phoenix，还是两者都接。
3. contacts 的 source vocabulary / source hierarchy 是否要提前对齐未来 CRM/POS integration。
4. Task 页面 source 联动先做独立 /sources endpoint，还是直接扩 timeline endpoint。
5. rejected proposal 的 source reverse lookup 是否需要高频 API。如果需要，后续加 source-edge projection。
6. production 前 retention / archive / partition policy 怎么定。

#Appendix A：目标 Schema 草案

这里的 ai_run_id 先按 soft join key 处理，不在 Phase 1 加硬 FK。原因和现有 task_sources.source_id、contact_timeline.ai_run_id 类似：这些账本行可能由不同 writer、不同重试路径、不同到达顺序写入。审计账本的第一原则是“先别丢记录”；如果 ai_runs 行因为 usage 写入失败或顺序问题暂时缺失，task_decision_attempts / call_analysis_runs 仍然应该能落库。等 writer 收敛到同一个事务边界，再评估是否加 deferred FK。

#A.1 task_decision_attempts

CREATE TABLE task_decision_attempts (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  idempotency_key text NOT NULL,

  store_id text NOT NULL,
  tenant_id uuid NULL,
  contact_phone text NOT NULL,
  contact_analysis_run_id uuid NOT NULL,
  decision_index integer NOT NULL,
  proposal_hash text NOT NULL,

  origin text NOT NULL,
  action text NOT NULL,
  type_category text NULL,

  target_task_ref text NULL,
  target_task_id uuid NULL,

  proposal_json jsonb NOT NULL,
  source_refs jsonb NULL,
  resolved_sources jsonb NULL,

  decision_status text NOT NULL,
  reject_reason text NULL,
  skip_reason text NULL,

  write_status text NOT NULL,
  written_task_id uuid NULL,
  zero_rows_reason text NULL,
  error_message text NULL,

  ai_run_id text NULL,

  created_at timestamptz NOT NULL DEFAULT now(),

  UNIQUE (idempotency_key, proposal_hash),

  CONSTRAINT chk_tda_origin
    CHECK (origin IN ('model','deterministic_fallback','code_derived')),
  CONSTRAINT chk_tda_action
    CHECK (action IN ('create_open','create_closed','close','update','record_progress','reopen')),
  CONSTRAINT chk_tda_decision_status
    CHECK (decision_status IN ('allow','reject','needs_review','skipped')),
  CONSTRAINT chk_tda_write_status
    CHECK (write_status IN ('not_attempted','written','zero_rows','error')),
  CONSTRAINT chk_tda_written_task_when_written
    CHECK ((write_status = 'written') = (written_task_id IS NOT NULL)),
  CONSTRAINT chk_tda_reject_reason_shape
    CHECK (decision_status != 'reject' OR reject_reason IS NOT NULL),
  CONSTRAINT chk_tda_skip_reason_shape
    CHECK (decision_status != 'skipped' OR skip_reason IS NOT NULL),
  CONSTRAINT chk_tda_zero_rows_reason_shape
    CHECK (write_status != 'zero_rows' OR zero_rows_reason IS NOT NULL),
  CONSTRAINT chk_tda_error_message_shape
    CHECK (write_status != 'error' OR error_message IS NOT NULL)
);

CREATE INDEX idx_task_decision_attempts_store_created
  ON task_decision_attempts (store_id, created_at DESC);

CREATE INDEX idx_task_decision_attempts_contact_run
  ON task_decision_attempts (contact_phone, store_id, contact_analysis_run_id);

CREATE INDEX idx_task_decision_attempts_guard
  ON task_decision_attempts (store_id, decision_status, reject_reason, created_at DESC);

CREATE INDEX idx_task_decision_attempts_written_task
  ON task_decision_attempts (written_task_id)
  WHERE written_task_id IS NOT NULL;

CREATE INDEX idx_task_decision_attempts_ai_run
  ON task_decision_attempts (ai_run_id)
  WHERE ai_run_id IS NOT NULL;

#A.2 ai_runs

CREATE TABLE ai_runs (
  run_id text PRIMARY KEY,

  stage text NOT NULL,
  store_id text NULL,
  tenant_id uuid NULL,
  contact_phone text NULL,
  telephony_session_id text NULL,

  provider text NULL,
  model text NULL,
  openrouter_id text NULL,
  prompt_version text NULL,
  prompt_hash text NULL,

  input_tokens integer NULL,
  output_tokens integer NULL,
  reasoning_tokens integer NULL,
  cache_read_tokens integer NULL,
  cache_write_tokens integer NULL,
  total_tokens integer NULL,
  estimated_cost_usd numeric NULL,
  latency_ms integer NULL,
  attempt_count integer NULL,

  status text NOT NULL,
  finish_reason text NULL,
  error_message text NULL,
  external_trace_id text NULL,
  prompt_s3_path text NULL,
  output_s3_path text NULL,

  created_at timestamptz NOT NULL DEFAULT now(),

  CONSTRAINT chk_ai_runs_status
    CHECK (status IN ('success','failed','aborted','validation_failed')),
  CONSTRAINT chk_ai_runs_error_message_shape
    CHECK (status NOT IN ('failed','validation_failed') OR error_message IS NOT NULL)
);

#A.3 call_analysis_runs

CREATE TABLE call_analysis_runs (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  telephony_session_id text NOT NULL,
  store_id text NOT NULL,
  tenant_id uuid NULL,

  ai_run_id text NOT NULL,

  status text NOT NULL,
  verdict_json jsonb NULL,
  verdict_hash text NULL,
  analysis_s3_path text NULL,
  error_message text NULL,

  created_at timestamptz NOT NULL DEFAULT now(),

  CONSTRAINT chk_car_status
    CHECK (status IN ('completed','failed','skipped')),
  CONSTRAINT chk_car_error_message_shape
    CHECK (status != 'failed' OR error_message IS NOT NULL)
);

CREATE INDEX idx_call_analysis_runs_call_created
  ON call_analysis_runs (telephony_session_id, created_at DESC);

CREATE INDEX idx_call_analysis_runs_store_created
  ON call_analysis_runs (store_id, created_at DESC);

CREATE UNIQUE INDEX uq_call_analysis_runs_ai_run
  ON call_analysis_runs (ai_run_id);

#A.4 calls latest provenance

ALTER TABLE calls ADD COLUMN ai_run_id text NULL;
ALTER TABLE calls ADD COLUMN ai_prompt_version text NULL;
ALTER TABLE calls ADD COLUMN ai_prompt_hash text NULL;

#A.5 contacts source / verified fields

ALTER TABLE contacts ADD COLUMN lifecycle_stage_source text NULL;
ALTER TABLE contacts ADD COLUMN lifecycle_stage_verified_at timestamptz NULL;
ALTER TABLE contacts ADD COLUMN lifecycle_stage_verified_by text NULL;
ALTER TABLE contacts ADD COLUMN lifecycle_stage_trust_score smallint NOT NULL DEFAULT 0;

ALTER TABLE contacts ADD COLUMN lead_status_source text NULL;
ALTER TABLE contacts ADD COLUMN lead_status_verified_at timestamptz NULL;
ALTER TABLE contacts ADD COLUMN lead_status_verified_by text NULL;
ALTER TABLE contacts ADD COLUMN lead_status_trust_score smallint NOT NULL DEFAULT 0;

ALTER TABLE contacts ADD COLUMN do_not_contact_source text NULL;
ALTER TABLE contacts ADD COLUMN do_not_contact_verified_at timestamptz NULL;
ALTER TABLE contacts ADD COLUMN do_not_contact_verified_by text NULL;
ALTER TABLE contacts ADD COLUMN do_not_contact_trust_score smallint NOT NULL DEFAULT 0;

#A.6 fieldChanges metadata shape

{
  "fieldChanges": [
    {
      "field": "lifecycleStage",
      "old": "lead",
      "new": "member",
      "source": "ai_inferred",
      "evidenceRefs": ["C1"],
      "reason": "Customer discussed active membership upgrade options."
    }
  ],
  "taskDecisionCounts": {
    "proposed": 5,
    "allow": 3,
    "reject": 1,
    "skipped": 1,
    "zero_rows": 0
  }
}

#Appendix B：代码证据索引

• callytics-common/src/db/schema/task-sources.ts: task_sources 已存在，source type / role / unique index / source lookup index 已定义。
• callytics-common/src/db/schema/task-progress-events.ts: task_progress_events 是 task 创建后的 append-only progress stream。
• callytics-common/src/db/schema/contact-timeline.ts: contact_timeline 已有 old/new/metadata 和 AI forensic typed columns。
• callytics-common/src/db/schema/calls.ts: calls 有 AI result、token/cost、ai_model_used，但没有 ai_run_id / ai_prompt_version / ai_prompt_hash。
• callytics-common/src/db/schema/contacts.ts: contacts 有 first_name_trust_score 和 do_not_contact_updated_by，但 lifecycle/leadStatus 缺 source/verified fields。
• callytics-common/src/domain/types.ts: ApplyResult 区分 allow / reject / needs_review，并有 RejectReason 与 zero-row reason。
• contacts-analyzer/src/core/task-refs.ts: resolveSourceRefs() 把 C1/M1/L1 解析成 call/message/lead source id。
• contacts-analyzer/src/infrastructure/neon-repository.ts: accepted task mutation 后已写 task_sources，zero-row mutation 会跳过 source write。
• ai-analysis-processor/src/infrastructure/neon-repository.ts: call_analysis.completed 已按 callId + analysisRunId 写入 contact_timeline，并带 old/new AI verdict。
• studio-website-monorepo/apps/api/src/routes/tasks/timeline.ts: task timeline 当前只组合 tasks + task_progress_events。
• studio-website-monorepo/apps/api/src/routes/tasks/communications.ts: communications 当前按 contact 拉 calls/messages，不按 task_sources 精准读取来源。