fix: placement verification and sidechat policy enforcement
- dm.py: placement-aware verify (verify_placement), checked_uuid logging, placement_mismatch events - main-chat-watchdog.py: P0 alerts on placement_mismatch - box-ctl.py: notify routes to sidechat, box policy command - jobs: ops-audit and pipe-demo use sidechats - response-harvester.py: chain deduplication Session: sidechat/ops-restore
This commit is contained in:
@@ -119,6 +119,18 @@ def append_jsonl(path, record):
|
||||
f.write(json.dumps(record) + "\n")
|
||||
|
||||
|
||||
# Case-insensitive failure/decline detection for [RESULT] text.
|
||||
# "declined"/"decline"/"reject" count as failures so declines do NOT
|
||||
# trigger onward pipeline chaining (previously only uppercase
|
||||
# FAILED/UNABLE/FAIL matched, so a lowercase "declined" was logged as success).
|
||||
FAIL_PREFIXES = ("failed", "unable", "fail", "declined", "decline", "reject", "error")
|
||||
|
||||
|
||||
def is_fail_result(result_text):
|
||||
t = (result_text or "").lstrip().lower()
|
||||
return t.startswith(FAIL_PREFIXES)
|
||||
|
||||
|
||||
def get_monitored_threads(target_agent=None):
|
||||
"""
|
||||
Build dict of threads to monitor per agent:
|
||||
@@ -342,10 +354,12 @@ def harvest_main_feed(cdp, agent, watermarks, followups, dry_run=False):
|
||||
|
||||
if author == "assistant":
|
||||
m_res = re.search(r"\[RESULT\s+([A-Za-z0-9_-]+)\]\s*(.*)", text, re.S)
|
||||
result_job_id = None
|
||||
if m_res:
|
||||
job_id = m_res.group(1).strip()
|
||||
result_job_id = job_id
|
||||
result_text = m_res.group(2).strip()
|
||||
is_fail = result_text.startswith("FAILED") or result_text.startswith("UNABLE") or result_text.startswith("FAIL")
|
||||
is_fail = is_fail_result(result_text)
|
||||
job_results += 1
|
||||
job_record = {
|
||||
"ts": utcnow(),
|
||||
@@ -361,7 +375,8 @@ def harvest_main_feed(cdp, agent, watermarks, followups, dry_run=False):
|
||||
append_jsonl(JOB_LOG, job_record)
|
||||
trigger_chain_next(job_id, result_text, success=not is_fail)
|
||||
|
||||
clear_matching_followups(followups, agent, "main", mid, text, dry_run)
|
||||
clear_matching_followups(followups, agent, "main", mid, text, dry_run,
|
||||
job_id=result_job_id)
|
||||
|
||||
return new_messages, new_wm, job_results
|
||||
|
||||
@@ -462,10 +477,12 @@ def harvest_agent_thread(cdp, agent, thread_info, watermarks, followups, dry_run
|
||||
# Check for [RESULT <job_id>] in assistant messages
|
||||
if author == "assistant":
|
||||
m_res = re.search(r"\[RESULT\s+([A-Za-z0-9_-]+)\]\s*(.*)", text, re.S)
|
||||
result_job_id = None
|
||||
if m_res:
|
||||
job_id = m_res.group(1).strip()
|
||||
result_job_id = job_id
|
||||
result_text = m_res.group(2).strip()
|
||||
is_fail = result_text.startswith("FAILED") or result_text.startswith("UNABLE") or result_text.startswith("FAIL")
|
||||
is_fail = is_fail_result(result_text)
|
||||
job_results += 1
|
||||
|
||||
job_record = {
|
||||
@@ -484,13 +501,42 @@ def harvest_agent_thread(cdp, agent, thread_info, watermarks, followups, dry_run
|
||||
trigger_chain_next(job_id, result_text, success=not is_fail)
|
||||
|
||||
# Check and clear pending follow-ups
|
||||
clear_matching_followups(followups, agent, thread_id, mid, text, dry_run)
|
||||
clear_matching_followups(followups, agent, thread_id, mid, text, dry_run,
|
||||
job_id=result_job_id)
|
||||
|
||||
return new_messages, new_wm, job_results
|
||||
|
||||
|
||||
# Chain deduplication
|
||||
CHAINED_JOBS_FILE = Path(__file__).parent / "chained-jobs.json"
|
||||
|
||||
def _has_chained(job_id):
|
||||
try:
|
||||
if CHAINED_JOBS_FILE.exists():
|
||||
import json as _j
|
||||
with open(CHAINED_JOBS_FILE) as f:
|
||||
return job_id in _j.load(f)
|
||||
except: pass
|
||||
return False
|
||||
|
||||
def _mark_chained(job_id):
|
||||
try:
|
||||
import json as _j
|
||||
c = []
|
||||
if CHAINED_JOBS_FILE.exists():
|
||||
with open(CHAINED_JOBS_FILE) as f: c = _j.load(f)
|
||||
if job_id not in c:
|
||||
c.append(job_id)
|
||||
c = c[-1000:]
|
||||
with open(CHAINED_JOBS_FILE, "w") as f: _j.dump(c, f)
|
||||
except: pass
|
||||
|
||||
def trigger_chain_next(job_id, result_text, success=True):
|
||||
"""If the completed job has on_success, on_failure, or chain_next, dispatch downstream."""
|
||||
if _has_chained(job_id):
|
||||
return
|
||||
_mark_chained(job_id)
|
||||
|
||||
m = re.match(r"^(.*)-(\d{8}-\d{6}-[a-f0-9]{8})$", job_id)
|
||||
if m:
|
||||
job_name = m.group(1)
|
||||
@@ -557,8 +603,14 @@ def trigger_chain_next(job_id, result_text, success=True):
|
||||
pass
|
||||
|
||||
|
||||
def clear_matching_followups(followups, agent, thread_id, mid, text, dry_run=False):
|
||||
"""Resolve follow-up records if an assistant message is detected in the thread."""
|
||||
def clear_matching_followups(followups, agent, thread_id, mid, text, dry_run=False,
|
||||
job_id=None):
|
||||
"""Resolve follow-up records if an assistant message is detected in the thread.
|
||||
|
||||
Matches on thread identity (thread_uuid or target='main') OR on job_id
|
||||
(from a [RESULT <job_id>] reply). The job_id path works regardless of
|
||||
thread_uuid or target, fixing ghost followups with null thread_uuid.
|
||||
"""
|
||||
if not followups:
|
||||
return
|
||||
|
||||
@@ -576,7 +628,13 @@ def clear_matching_followups(followups, agent, thread_id, mid, text, dry_run=Fal
|
||||
elif f_rec.get("thread_uuid") and f_rec.get("thread_uuid") == thread_id:
|
||||
match_thread = True
|
||||
|
||||
if match_thread:
|
||||
# Match by job_id (from [RESULT <job_id>]) -- works regardless of
|
||||
# thread_uuid or target. This is an ADDITIONAL path, not a replacement.
|
||||
match_job = False
|
||||
if job_id and f_rec.get("job_id") and f_rec.get("job_id") == job_id:
|
||||
match_job = True
|
||||
|
||||
if match_thread or match_job:
|
||||
f_rec["status"] = "resolved"
|
||||
f_rec["resolved_at"] = utcnow()
|
||||
f_rec["resolved_by_mid"] = mid
|
||||
|
||||
Reference in New Issue
Block a user