核心功能: 本技能提供器v1等能力。

工作流检查点管理 v1.0 (ARCH-5 PG为主+SQLite为缓存)

PG为唯一权威源,SQLite仅本地缓存可丢失,崩溃后从PG重建。取消v7.0双写Saga,简化一致性保证。

使用场景

1. 工作流每步执行后保存检查点(崩溃可恢复) 2. 工作流重启时从PG读取最后检查点恢复 3. 高频读取用SQLite缓存加速(缓存未命中自动回退PG) 4. SQLite损坏/丢失后从PG重建缓存 5. 定期一致性验证(比对PG与SQLite) 6. 多租户检查点隔离与按租户重建 7. 健康检查与运维监控

核心概念

PG权威源: workflow_checkpoints表,UNIQUE(workflow_id, step_id),所有检查点以PG为准。写入用INSERT ON CONFLICT DO UPDATE(upsert)。

SQLite缓存: data/checkpoint_mcp/cache.db,checkpoint_cache表,PRIMARY KEY(workflow_id, step_id)。可丢失,失败不影响业务。WAL模式提升并发。

写入路径: state_data → PG(同步,权威源) → SQLite(异步,缓存,可丢失)。PG成功即返回成功。

读取路径: SQLite(快速) → 未命中/过期 → PG(权威源) → 回填SQLite。缓存TTL默认300秒。

不再双写: 取消v7.0 Saga补偿事务+event sourcing replay,PG单点负责一致性。

工作流

流程A: 保存检查点(save_checkpoint)

1. 调用checkpoint-mcp的save_checkpoint(workflow_id, step_id, state_data, tenant_id)
2. PG同步写入(INSERT ON CONFLICT DO UPDATE),返回id/created_at/updated_at
3. SQLite异步缓存(fire-and-forget,失败不影响业务)
4. 返回success=true + authority_source="pg" + cache_written

流程B: 从PG权威源读取(get_checkpoint)

1. 调用get_checkpoint(workflow_id, step_id)
2. 直接从PG读取,保证数据一致性(崩溃恢复/最终确认场景)
3. 返回checkpoint完整信息(id/tenant_id/state_data/created_at/updated_at)
4. 不存在时返回found=false(非错误)

流程C: 快速读取(get_cached_state)

1. 调用get_cached_state(workflow_id, step_id)
2. 优先从SQLite缓存读取(快速),命中且未过期(TTL内)返回source="sqlite"
3. 缓存未命中/过期 → 从PG权威源读取,返回source="pg" + backfilled=true
4. 异步回填SQLite缓存(下次读取命中)
5. PG和SQLite均无 → 返回found=false + source="none"

流程D: 列出工作流检查点(list_checkpoints)

1. 调用list_checkpoints(workflow_id, tenant_id="")
2. 从PG权威源列出该工作流所有检查点(按updated_at降序)
3. tenant_id非空时按租户过滤
4. 返回checkpoints列表 + count

流程E: 显式缓存(cache_to_sqlite)

1. 调用cache_to_sqlite(workflow_id, step_id, state_data)
2. 同步写入SQLite缓存(不写PG,仅缓存场景)
3. 返回written=true/false + cached_at + lossy=true

流程F: 崩溃恢复重建(rebuild_sqlite_cache)

1. SQLite丢失/损坏时调用rebuild_sqlite_cache(tenant_id="default")
2. 从PG读取该租户所有检查点(最多10000条)
3. 清空SQLite中相关workflow的缓存条目
4. 批量写入SQLite缓存
5. 返回pg_read_count/deleted_before/rebuilt_count/elapsed_sec

流程G: 一致性验证(verify_checkpoint_integrity)

1. 调用verify_checkpoint_integrity(workflow_id="")
2. 从PG和SQLite分别读取检查点
3. 比对差异:cache_only(SQLite有PG无)/pg_only(PG有SQLite无)/data_mismatch(数据不同)
4. 生成报告(原子写入reports目录)
5. 返回consistent布尔 + mismatches详情 + report_path

异常处理

输入格式

{
  "action": "save|get|list|cache|cached_get|rebuild|verify|healthcheck",
  "workflow_id": "wf_20260707_001",
  "step_id": "step_collect_materials",
  "tenant_id": "default",
  "state_data": {"progress": 50, "last_url": "https://...", "items": [1,2,3]}
}

字段说明:

• action: 操作类型(save保存/get PG读取/list列表/cache显式缓存/cached_get快速读取/rebuild重建/verify验证/healthcheck健康检查)
• workflow_id: 工作流ID(除healthcheck外必填)
• step_id: 步骤ID(save/get/cache/cached_get必填)
• tenant_id: 租户ID(默认"default",save/list/rebuild使用)
• state_data: 状态数据(save/cache必填,任意JSON结构)

输出格式

{
  "success": true,
  "data": {
    "workflow_id": "wf_20260707_001",
    "step_id": "step_collect_materials",
    "tenant_id": "default",
    "id": 42,
    "created_at": "2026-07-07T10:00:00+08:00",
    "updated_at": "2026-07-07T10:05:30+08:00",
    "cache_written": true,
    "authority_source": "pg"
  },
  "error": null,
  "code": null
}

字段说明:

• authority_source: "pg"(权威源)/"sqlite"(缓存命中)/"none"(不存在)
• cache_written: SQLite缓存是否写入成功(可丢失语义)
• backfilled: 缓存未命中时是否从PG回填SQLite(快速读取场景)
• consistent: 一致性验证是否通过(verify场景)
• lossy: SQLite缓存是否可丢失(true)

示例

示例1: 保存检查点并快速读取

1. 调用save_checkpoint(workflow_id="wf_001", step_id="step_1", state_data={"progress":30}, tenant_id="default")
2. 返回: {success:true, data:{id:1, authority_source:"pg", cache_written:true}}
3. 调用get_cached_state(workflow_id="wf_001", step_id="step_1")
4. 返回: {success:true, data:{found:true, source:"sqlite", state_data:{"progress":30}, backfilled:false}}

示例2: 缓存未命中回退PG

1. 删除SQLite缓存文件(模拟崩溃)
2. 调用get_cached_state(workflow_id="wf_001", step_id="step_1")
3. SQLite未命中 → 从PG读取 → 返回source="pg" + backfilled=true
4. 再次调用get_cached_state → 命中SQLite,返回source="sqlite"

示例3: 崩溃恢复重建

1. SQLite文件损坏/丢失
2. 调用rebuild_sqlite_cache(tenant_id="default")
3. 从PG读取100条检查点,清空旧缓存,重建100条
4. 返回: {success:true, data:{pg_read_count:100, deleted_before:0, rebuilt_count:100, elapsed_sec:0.5}}

示例4: 一致性验证

1. 调用verify_checkpoint_integrity(workflow_id="wf_001")
2. 比对PG(5条)与SQLite(4条),发现pg_only=1条
3. 返回: {success:true, data:{consistent:false, pg_count:5, sqlite_count:4, mismatches:{pg_only:[{workflow_id:"wf_001",step_id:"step_3"}]}}}
4. 调用rebuild_sqlite_cache修复不一致

验证标准

变更历史