host
be2962525b
- peer_messages.row_get gains default=None (was None-only); the three old variants agree on all real row types (None / dict / sqlite3.Row): auto_memory's old default was already None, and lifecycle's old None special-case and dict .get give identical results on those types, so no call-site behavior changes and no call site passes a non-None default. - auto_memory and lifecycle delete their private _row_get copies and import peer_messages.row_get as _row_get (precedent: server, worker, handler_project, handler_support already do this); peer_messages imports no cloud modules, so no import cycle. lifecycle's old broad-Exception catch differs only for pathological __getitem__ inputs, which are not DB rows and are out of contract; this is documented on row_get, not left as a second story. - Scope note: other modules' own row_get copies (approvals, automations, marker_tools, todos, wait_ops) are out of this item's stated scope and are untouched. - Also drops auto_memory's unused 'import os' (pre-existing F401), so touched files are ruff E,F clean. - Tests: cloud/tests/test_row_get_unified.py locks the identity (both aliases ARE row_get), the real-row-type matrix, the None-default fallback, and snapshot_to_dict via the unified path. (cherry picked from commit 1644a0b837bfefb13cdc3c93d87fc1b246026607)
285 lines
10 KiB
Python
285 lines
10 KiB
Python
"""线程生命周期补充：worktree 快照、任务外部引用、自动化心跳绑定。
|
||
|
||
本模块只补控制面已有能力的缺口，不发明执行能力：
|
||
|
||
- worktree 快照（task_worktree_snapshots）：归档线程时真正要
|
||
保住的是“未提交修改 + 未推送 commit 的位置 + 非忽略文件
|
||
清单”。快照记 branch / head_sha / git status（截 4000 字）/
|
||
git diff（截 20000 字）/ untracked 路径清单。注意诚实边界：
|
||
untracked 文件只记路径不拷内容，diff 只覆盖已跟踪文件的
|
||
修改；快照是事后可查的证据，不是可一键还原的备份。恢复
|
||
（restore）只在 worktree 目录还在时解归档；目录已删时
|
||
回 409 并附快照（重建需要仓库路径，服务端没有，不编造）。
|
||
- 任务外部引用（task_refs）：任务附件表只记文件包路径，
|
||
交付物里的“生成的文档 / 构建结果 / 外部服务结果引用”
|
||
没有地方放。refs 只记 kind + ref + note 的引用行，不存
|
||
文件内容、不校验外部链接可达——引用就是引用。
|
||
- 自动化心跳绑定（automation_bindings）：自动化定义本身
|
||
只有 interval + prompt，没有“绑到哪个线程”。绑定记
|
||
automation -> task + mode（heartbeat | cron）。heartbeat
|
||
是“到期唤醒已有线程”（走 thread/message 的投递路）；
|
||
cron 是“到期独立跑新任务”（走自动化自身的产出路）。
|
||
本模块只记绑定与取绑定；到期触发仍由外部触发器
|
||
（host run-due / 未来的 run_due_automation）调用
|
||
/heartbeat 端点，服务端没有也不会偷偷起定时器。
|
||
|
||
三张表都走 UNMIGRATED_TABLES（见 cloud/migrations.py）：
|
||
运行时 ensure 建表，与 task_handoffs 同款，不进
|
||
POSTGRES_SCHEMA 快照。"""
|
||
|
||
import json
|
||
import time
|
||
import uuid
|
||
|
||
# 行取值统一走 peer_messages.row_get（A7）：本模块旧私有版在
|
||
# 真实行类型（None / dict / sqlite3.Row）下与之结果完全一致
|
||
# （None 特判、dict .get、缺列回 None），调用处仍用 _row_get
|
||
# 别名，行为不变；旧版宽 Exception 兜底的病态行不在契约内，
|
||
# 见 peer_messages.row_get docstring。
|
||
from cloud.peer_messages import row_get as _row_get
|
||
|
||
SNAPSHOT_STATUS_LIMIT = 4000
|
||
SNAPSHOT_DIFF_LIMIT = 20000
|
||
REF_LIMIT = 2000
|
||
NOTE_LIMIT = 1000
|
||
REF_KINDS = ("doc", "build", "external", "pr", "file")
|
||
BINDING_MODES = ("heartbeat", "cron")
|
||
|
||
|
||
# -- worktree 快照 ------------------------------------------------------
|
||
|
||
def ensure_snapshots_table(conn):
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS task_worktree_snapshots("
|
||
"id TEXT PRIMARY KEY, "
|
||
"task_id TEXT, "
|
||
"project_id TEXT, "
|
||
"worktree_path TEXT, "
|
||
"branch TEXT, "
|
||
"head_sha TEXT, "
|
||
"status_text TEXT, "
|
||
"diff_text TEXT, "
|
||
"untracked_json TEXT, "
|
||
"created_at DOUBLE PRECISION, "
|
||
"restored_at DOUBLE PRECISION, "
|
||
"restore_note TEXT)")
|
||
conn.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_snapshots_task"
|
||
" ON task_worktree_snapshots(task_id, created_at)")
|
||
|
||
|
||
def snapshot_to_dict(row):
|
||
try:
|
||
untracked = json.loads(_row_get(row, "untracked_json") or "[]")
|
||
except ValueError:
|
||
untracked = []
|
||
if not isinstance(untracked, list):
|
||
untracked = []
|
||
return {
|
||
"id": _row_get(row, "id"),
|
||
"task_id": _row_get(row, "task_id"),
|
||
"project_id": _row_get(row, "project_id"),
|
||
"worktree_path": _row_get(row, "worktree_path"),
|
||
"branch": _row_get(row, "branch"),
|
||
"head_sha": _row_get(row, "head_sha"),
|
||
"status_text": _row_get(row, "status_text"),
|
||
"diff_text": _row_get(row, "diff_text"),
|
||
"untracked": untracked,
|
||
"created_at": _row_get(row, "created_at"),
|
||
"restored_at": _row_get(row, "restored_at"),
|
||
"restore_note": _row_get(row, "restore_note"),
|
||
}
|
||
|
||
|
||
def create_snapshot(conn, task_id, project_id, worktree_path, branch,
|
||
head_sha, status_text, diff_text, untracked, now=None):
|
||
"""插一条快照并提交。status/diff 在调用方截断，这里再截
|
||
一次防直接调用方越界；untracked 只接受 list[str]。"""
|
||
ensure_snapshots_table(conn)
|
||
snap_id = uuid.uuid4().hex
|
||
now = time.time() if now is None else now
|
||
untracked = [str(p) for p in (untracked or []) if isinstance(p, str)]
|
||
conn.execute(
|
||
"INSERT INTO task_worktree_snapshots(id, task_id, project_id,"
|
||
" worktree_path, branch, head_sha, status_text, diff_text,"
|
||
" untracked_json, created_at, restored_at, restore_note)"
|
||
" VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL)",
|
||
(snap_id, task_id, project_id, worktree_path, branch,
|
||
head_sha, (status_text or "")[:SNAPSHOT_STATUS_LIMIT],
|
||
(diff_text or "")[:SNAPSHOT_DIFF_LIMIT],
|
||
json.dumps(untracked), now))
|
||
conn.commit()
|
||
row = conn.execute(
|
||
"SELECT * FROM task_worktree_snapshots WHERE id = ?",
|
||
(snap_id,)).fetchone()
|
||
return snapshot_to_dict(row)
|
||
|
||
|
||
def list_snapshots(conn, task_id):
|
||
ensure_snapshots_table(conn)
|
||
rows = conn.execute(
|
||
"SELECT * FROM task_worktree_snapshots WHERE task_id = ?"
|
||
" ORDER BY created_at DESC, id",
|
||
(task_id,)).fetchall()
|
||
return [snapshot_to_dict(r) for r in rows]
|
||
|
||
|
||
def latest_snapshot(conn, task_id):
|
||
rows = list_snapshots(conn, task_id)
|
||
return rows[0] if rows else None
|
||
|
||
|
||
def mark_restored(conn, snapshot_id, note, now=None):
|
||
ensure_snapshots_table(conn)
|
||
now = time.time() if now is None else now
|
||
cur = conn.execute(
|
||
"UPDATE task_worktree_snapshots SET restored_at = ?,"
|
||
" restore_note = ? WHERE id = ?",
|
||
(now, note, snapshot_id))
|
||
conn.commit()
|
||
if cur.rowcount != 1:
|
||
return None
|
||
row = conn.execute(
|
||
"SELECT * FROM task_worktree_snapshots WHERE id = ?",
|
||
(snapshot_id,)).fetchone()
|
||
return snapshot_to_dict(row)
|
||
|
||
|
||
# -- 任务外部引用 ---------------------------------------------------------
|
||
|
||
def ensure_refs_table(conn):
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS task_refs("
|
||
"id TEXT PRIMARY KEY, "
|
||
"task_id TEXT, "
|
||
"project_id TEXT, "
|
||
"kind TEXT, "
|
||
"ref TEXT, "
|
||
"note TEXT, "
|
||
"created_by TEXT, "
|
||
"created_at DOUBLE PRECISION)")
|
||
conn.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_refs_task"
|
||
" ON task_refs(task_id, created_at)")
|
||
|
||
|
||
def ref_to_dict(row):
|
||
return {
|
||
"id": _row_get(row, "id"),
|
||
"task_id": _row_get(row, "task_id"),
|
||
"project_id": _row_get(row, "project_id"),
|
||
"kind": _row_get(row, "kind"),
|
||
"ref": _row_get(row, "ref"),
|
||
"note": _row_get(row, "note"),
|
||
"created_by": _row_get(row, "created_by"),
|
||
"created_at": _row_get(row, "created_at"),
|
||
}
|
||
|
||
|
||
def add_ref(conn, task_id, project_id, kind, ref, note, created_by,
|
||
now=None):
|
||
"""校验后插引用行。kind 不在白名单、ref 为空或超长、
|
||
note 超长时抛 ValueError（调用方转 400）。"""
|
||
if kind not in REF_KINDS:
|
||
raise ValueError("kind must be one of %s" % ", ".join(REF_KINDS))
|
||
if not ref or not isinstance(ref, str) or not ref.strip():
|
||
raise ValueError("ref required")
|
||
ref = ref.strip()
|
||
if len(ref) > REF_LIMIT:
|
||
raise ValueError("ref too long")
|
||
if note is not None and len(str(note)) > NOTE_LIMIT:
|
||
raise ValueError("note too long")
|
||
ensure_refs_table(conn)
|
||
ref_id = uuid.uuid4().hex
|
||
now = time.time() if now is None else now
|
||
conn.execute(
|
||
"INSERT INTO task_refs(id, task_id, project_id, kind, ref,"
|
||
" note, created_by, created_at) VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(ref_id, task_id, project_id, kind, ref, note, created_by,
|
||
now))
|
||
conn.commit()
|
||
row = conn.execute(
|
||
"SELECT * FROM task_refs WHERE id = ?", (ref_id,)).fetchone()
|
||
return ref_to_dict(row)
|
||
|
||
|
||
def list_refs(conn, task_id):
|
||
ensure_refs_table(conn)
|
||
rows = conn.execute(
|
||
"SELECT * FROM task_refs WHERE task_id = ?"
|
||
" ORDER BY created_at, id",
|
||
(task_id,)).fetchall()
|
||
return [ref_to_dict(r) for r in rows]
|
||
|
||
|
||
def delete_ref(conn, task_id, ref_id):
|
||
"""只删属于该任务的引用行；行不存在或不属于该任务时
|
||
返回 False（调用方一律 404，不泄露跨任务存在性）。"""
|
||
ensure_refs_table(conn)
|
||
cur = conn.execute(
|
||
"DELETE FROM task_refs WHERE id = ? AND task_id = ?",
|
||
(ref_id, task_id))
|
||
conn.commit()
|
||
return cur.rowcount == 1
|
||
|
||
|
||
# -- 自动化心跳绑定 --------------------------------------------------------
|
||
|
||
def ensure_bindings_table(conn):
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS automation_bindings("
|
||
"automation_id TEXT PRIMARY KEY, "
|
||
"project_id TEXT, "
|
||
"task_id TEXT, "
|
||
"mode TEXT, "
|
||
"created_at DOUBLE PRECISION)")
|
||
|
||
|
||
def binding_to_dict(row):
|
||
if row is None:
|
||
return None
|
||
return {
|
||
"automation_id": _row_get(row, "automation_id"),
|
||
"project_id": _row_get(row, "project_id"),
|
||
"task_id": _row_get(row, "task_id"),
|
||
"mode": _row_get(row, "mode"),
|
||
"created_at": _row_get(row, "created_at"),
|
||
}
|
||
|
||
|
||
def bind_automation(conn, automation_id, project_id, task_id, mode,
|
||
now=None):
|
||
if mode not in BINDING_MODES:
|
||
raise ValueError("mode must be heartbeat or cron")
|
||
ensure_bindings_table(conn)
|
||
now = time.time() if now is None else now
|
||
conn.execute(
|
||
"INSERT INTO automation_bindings(automation_id, project_id,"
|
||
" task_id, mode, created_at) VALUES(?, ?, ?, ?, ?)"
|
||
" ON CONFLICT(automation_id) DO UPDATE SET"
|
||
" task_id = excluded.task_id, mode = excluded.mode,"
|
||
" created_at = excluded.created_at",
|
||
(automation_id, project_id, task_id, mode, now))
|
||
conn.commit()
|
||
row = conn.execute(
|
||
"SELECT * FROM automation_bindings WHERE automation_id = ?",
|
||
(automation_id,)).fetchone()
|
||
return binding_to_dict(row)
|
||
|
||
|
||
def get_binding(conn, automation_id):
|
||
ensure_bindings_table(conn)
|
||
row = conn.execute(
|
||
"SELECT * FROM automation_bindings WHERE automation_id = ?",
|
||
(automation_id,)).fetchone()
|
||
return binding_to_dict(row)
|
||
|
||
|
||
def unbind_automation(conn, automation_id):
|
||
ensure_bindings_table(conn)
|
||
cur = conn.execute(
|
||
"DELETE FROM automation_bindings WHERE automation_id = ?",
|
||
(automation_id,))
|
||
conn.commit()
|
||
return cur.rowcount == 1
|