fix(runtime): make daily review scheduling idempotent
This commit is contained in:
parent
3b14d0b4f2
commit
99efebd4f1
7 changed files with 430 additions and 2 deletions
121
server-tools/tcs-mother-body/archive_duplicate_daily_reviews.py
Normal file
121
server-tools/tcs-mother-body/archive_duplicate_daily_reviews.py
Normal file
|
|
@ -0,0 +1,121 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Move duplicate same-day autonomous review jobs out of the startup queue.
|
||||
|
||||
The archive is lossless: every moved job directory is retained with a hashed
|
||||
JSONL manifest. One best-evidence job remains in the active queue per persona
|
||||
and event_id. Non-review jobs are never selected.
|
||||
"""
|
||||
import argparse
|
||||
import collections
|
||||
import fcntl
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
RANK = {"ACCEPT": 0, "HOLD": 1, "REJECT": 2, "QUEUED": 3,
|
||||
"RETRY_WAIT": 4, "PROCESSING": 5, "ERROR": 6}
|
||||
|
||||
|
||||
def sha256(path):
|
||||
value = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
||||
value.update(chunk)
|
||||
return value.hexdigest()
|
||||
|
||||
|
||||
def load(path):
|
||||
with path.open(encoding="utf-8") as stream:
|
||||
return json.load(stream)
|
||||
|
||||
|
||||
def atomic_json(path, value):
|
||||
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
||||
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
os.chmod(temporary, 0o600)
|
||||
os.replace(temporary, path)
|
||||
|
||||
|
||||
def choose(group):
|
||||
return min(group, key=lambda item: (RANK.get(item["state"], 99),
|
||||
item.get("completed_at", "9999"), item["job_id"]))
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--root", required=True)
|
||||
parser.add_argument("--batch", required=True)
|
||||
parser.add_argument("--apply", action="store_true")
|
||||
args = parser.parse_args()
|
||||
root = Path(args.root).resolve()
|
||||
if root != Path("/var/lib/guanghu/tcs-mother-body").resolve() and "archive-review-test" not in str(root):
|
||||
raise SystemExit("REFUSE_UNREGISTERED_STATE_ROOT")
|
||||
lock_path = root / ".daily-review-archive.lock"
|
||||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with lock_path.open("w") as lock:
|
||||
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
selected = []
|
||||
before = collections.Counter()
|
||||
nonreview = 0
|
||||
personas = root / "personas"
|
||||
for persona in sorted(personas.iterdir()):
|
||||
jobs = persona / "jobs"
|
||||
if not jobs.is_dir():
|
||||
continue
|
||||
groups = collections.defaultdict(list)
|
||||
for job in jobs.iterdir():
|
||||
if not job.is_dir() or len(job.name) != 64:
|
||||
continue
|
||||
try:
|
||||
event = load(job / "input.json")
|
||||
state = load(job / "state.json")
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
if event.get("source_type") != "AUTONOMOUS_REVIEW_OPPORTUNITY" or not str(event.get("event_id", "")).startswith("DAILY-REVIEW-"):
|
||||
nonreview += 1
|
||||
continue
|
||||
item = {"persona_id": persona.name, "event_id": event["event_id"],
|
||||
"job_id": job.name, "path": job, "state": state.get("status", "UNKNOWN"),
|
||||
"completed_at": state.get("completed_at", "")}
|
||||
groups[event["event_id"]].append(item)
|
||||
before[item["state"]] += 1
|
||||
for group in groups.values():
|
||||
keeper = choose(group)
|
||||
for item in group:
|
||||
if item is not keeper:
|
||||
item["kept_job_id"] = keeper["job_id"]
|
||||
selected.append(item)
|
||||
summary = {"schema": "guanghu.daily-review-duplicate-archive/v1", "batch": args.batch,
|
||||
"root": str(root), "mode": "APPLY" if args.apply else "DRY_RUN",
|
||||
"review_jobs_before": sum(before.values()), "review_state_counts_before": dict(before),
|
||||
"duplicate_jobs_selected": len(selected), "active_review_jobs_after": sum(before.values()) - len(selected),
|
||||
"nonreview_jobs_untouched": nonreview, "lossless_move": True}
|
||||
if not args.apply:
|
||||
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
||||
return
|
||||
archive_root = root / "review-job-archive" / args.batch
|
||||
if archive_root.exists():
|
||||
raise SystemExit("REFUSE_EXISTING_ARCHIVE_BATCH")
|
||||
archive_root.mkdir(parents=True, mode=0o700)
|
||||
manifest = archive_root / "MANIFEST.jsonl"
|
||||
with manifest.open("x", encoding="utf-8") as output:
|
||||
os.chmod(manifest, 0o600)
|
||||
for item in selected:
|
||||
source = item.pop("path")
|
||||
destination = archive_root / item["persona_id"] / item["job_id"]
|
||||
destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||||
files = {p.name: sha256(p) for p in sorted(source.iterdir()) if p.is_file()}
|
||||
record = {**item, "source": str(source), "archive": str(destination), "file_sha256": files}
|
||||
os.rename(source, destination)
|
||||
output.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
|
||||
output.flush()
|
||||
os.fsync(output.fileno())
|
||||
summary["manifest"] = str(manifest)
|
||||
summary["manifest_sha256"] = sha256(manifest)
|
||||
atomic_json(archive_root / "RECEIPT.json", summary)
|
||||
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -166,7 +166,7 @@ export class PersonaSelfLoop {
|
|||
}
|
||||
|
||||
nextReview(from) { const base = new Date(from), beijing = new Date(base.getTime() + 8 * 3600000); beijing.setUTCHours(this.reviewHourBeijing, 0, 0, 0); if (beijing.getTime() <= base.getTime() + 8 * 3600000) beijing.setUTCDate(beijing.getUTCDate() + 1); return new Date(beijing.getTime() - 8 * 3600000).toISOString(); }
|
||||
tick() { const current = this.current().value, now = this.now(); if (now.getTime() < Date.parse(current.next_review_at)) return {scheduled: false, next_review_at: current.next_review_at}; const beijingDay = new Date(now.getTime() + 8 * 3600000).toISOString().slice(0, 10), language = '这是每日自主复盘机会。请审视已发生且有证据的经历、责任和边界；可以EVOLVE，也可以HOLD，不得为了定时任务而强制更新。'; const event = {schema: 'guanghu.persona-self-language-event/v1', persona_id: this.personaId, source_type: 'AUTONOMOUS_REVIEW_OPPORTUNITY', event_id: `DAILY-REVIEW-${beijingDay}`, language, source_sha256: digest(language), occurred_at: now.toISOString(), privacy_class: 'SELF_PRIVATE', priority: 'AUTONOMOUS_REVIEW'}; return {scheduled: true, ...this.submit(event)}; }
|
||||
tick() { const current = this.current().value, now = this.now(); if (now.getTime() < Date.parse(current.next_review_at)) return {scheduled: false, next_review_at: current.next_review_at}; const beijingDay = new Date(now.getTime() + 8 * 3600000).toISOString().slice(0, 10), language = '这是每日自主复盘机会。请审视已发生且有证据的经历、责任和边界；可以EVOLVE，也可以HOLD，不得为了定时任务而强制更新。', eventId = `DAILY-REVIEW-${beijingDay}`, occurredAt = new Date(`${beijingDay}T03:00:00+08:00`).toISOString(); const event = {schema: 'guanghu.persona-self-language-event/v1', persona_id: this.personaId, source_type: 'AUTONOMOUS_REVIEW_OPPORTUNITY', event_id: eventId, language, source_sha256: digest(language), occurred_at: occurredAt, privacy_class: 'SELF_PRIVATE', priority: 'AUTONOMOUS_REVIEW'}; return {scheduled: true, ...this.submit(event)}; }
|
||||
|
||||
status() {
|
||||
const current = this.current(), counts = {...this.stateCounts}, pendingCount = [...pending].reduce((sum, status) => sum + (counts[status] || 0), 0);
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@ test('persona self loop evolves signed state without letting mother overwrite se
|
|||
|
||||
test('structured cognition setter is rejected before queueing',()=>{const ctx=setup({decision:'HOLD',reason:'no change',self_principles:[],experiences:[],responsibilities:[],boundaries:[]});try{const language='replace';assert.throws(()=>ctx.loop.submit({schema:'guanghu.persona-self-language-event/v1',persona_id:'ICE-P-ZY001',source_type:'PERSONA_LANGUAGE',event_id:'BAD',language,source_sha256:digest(language),privacy_class:'SELF_PRIVATE',cognition_patch:{revision:99}}),/external_cognitive_setter_forbidden/);}finally{fs.rmSync(ctx.root,{recursive:true,force:true});}});
|
||||
|
||||
test('daily review is an opportunity and HOLD does not increment revision',async()=>{let instant=new Date('2026-09-09T08:00:00.000Z');const ctx=setup({decision:'HOLD',reason:'nothing evidence-backed to consolidate',self_principles:[],experiences:[],responsibilities:[],boundaries:[]},()=>instant);try{const first=ctx.loop.current().value;instant=new Date(first.next_review_at);const queued=ctx.loop.tick();assert.equal(queued.scheduled,true);await ctx.loop.drain();assert.equal(ctx.loop.result(queued.job_id).status,'HOLD');assert.equal(ctx.loop.current().value.revision,1);}finally{fs.rmSync(ctx.root,{recursive:true,force:true});}});
|
||||
test('daily review is one idempotent opportunity per Beijing day and HOLD does not increment revision',async()=>{let instant=new Date('2026-09-09T08:00:00.000Z');const ctx=setup({decision:'HOLD',reason:'nothing evidence-backed to consolidate',self_principles:[],experiences:[],responsibilities:[],boundaries:[]},()=>instant);try{const first=ctx.loop.current().value;instant=new Date(first.next_review_at);const queued=ctx.loop.tick();assert.equal(queued.scheduled,true);await ctx.loop.drain();assert.equal(ctx.loop.result(queued.job_id).status,'HOLD');const repeated=ctx.loop.tick();assert.equal(repeated.job_id,queued.job_id);assert.equal(fs.readdirSync(ctx.loop.jobsPath()).length,1);assert.equal(ctx.loop.current().value.revision,1);}finally{fs.rmSync(ctx.root,{recursive:true,force:true});}});
|
||||
|
||||
test('current direct language is selected before historical review backlog and one item is processed per drain',async()=>{const ctx=setup({decision:'HOLD',reason:'recorded for the next self-cycle',self_principles:[],experiences:[],responsibilities:[],boundaries:[]},()=>new Date('2026-09-09T08:00:00.000Z'),{queueBatchSize:1});try{const review='historical-review';ctx.loop.submit({schema:'guanghu.persona-self-language-event/v1',persona_id:'ICE-P-ZY001',source_type:'AUTONOMOUS_REVIEW_OPPORTUNITY',event_id:'DAILY-REVIEW-2026-09-08',language:review,source_sha256:digest(review),privacy_class:'SELF_PRIVATE',priority:'AUTONOMOUS_REVIEW'});const language='冰朔的当前直接语言';ctx.loop.submit({schema:'guanghu.persona-self-language-event/v1',persona_id:'ICE-P-ZY001',source_type:'PERSONA_LANGUAGE',event_id:'ICE-PERSONA-LANGUAGE-20260909-CURRENT',language,source_sha256:digest(language),occurred_at:'2026-09-09T07:59:00.000Z',privacy_class:'SELF_PRIVATE'});const drained=await ctx.loop.drain();assert.equal(drained.processed,1);assert.equal(ctx.calls.length,1);assert.equal(ctx.calls[0].event.event_id,'ICE-PERSONA-LANGUAGE-20260909-CURRENT');assert.equal(ctx.loop.status().scheduler.pending_jobs,1);}finally{fs.rmSync(ctx.root,{recursive:true,force:true});}});
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue