"""Prime Agent threads skill: manage tracked commitment threads from Python."""
from __future__ import annotations
import json
import os
import pathlib
import time
import uuid
from typing import Any
def _get_session_path() -> pathlib.Path | None:
sess_dir = os.environ.get("RLM_SESSION_DIR")
if sess_dir:
sid = pathlib.Path(sess_dir).name
sp = pathlib.Path.home() / f".prime/agent/sessions/{sid}.jsonl"
if sp.exists():
return sp
sessions = sorted((pathlib.Path.home() / ".prime/agent/sessions").glob("*.jsonl"), key=lambda p: p.stat().st_mtime)
return sessions[-1] if sessions else None
def _append_event(event_data: dict[str, Any]) -> None:
sp = _get_session_path()
if not sp or not sp.exists():
raise RuntimeError("No active session file found")
entry = {
"type": "custom",
"customType": "commitment-event",
"data": event_data,
"id": str(uuid.uuid4())[:8],
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
with open(sp, "a", encoding="utf-8") as f:
f.write(json.dumps(entry) + "\n")
def _read_threads() -> list[dict[str, Any]]:
sp = _get_session_path()
if not sp or not sp.exists():
return []
events = []
with open(sp, "r", encoding="utf-8", errors="ignore") as f:
for line in f:
try:
row = json.loads(line)
if row.get("type") == "custom" and row.get("customType") == "commitment-event":
events.append(row["data"])
except Exception:
pass
threads = []
by_id = {}
for ev in events:
kind = ev.get("kind")
if kind == "observation":
out = ev.get("output", {})
for existing in out.get("existing", []):
tid = existing.get("id")
if tid in by_id:
th = by_id[tid]
for k in ["canonicalRequest", "requestedProperties", "implementedProperties", "contradictions", "evidenceActionIds", "status"]:
if k in existing and existing[k] is not None:
th[k] = existing[k]
th["updatedAt"] = ev.get("at", int(time.time() * 1000))
for add in out.get("additions", []):
tid = f"t{len(threads) + 1}"
new_th = {
"id": tid,
"canonicalRequest": add.get("canonicalRequest", ""),
"requestedProperties": add.get("requestedProperties", []),
"implementedProperties": add.get("implementedProperties", []),
"contradictions": add.get("contradictions", []),
"evidenceActionIds": add.get("evidenceActionIds", []),
"status": add.get("status", "open"),
"createdAt": ev.get("at", int(time.time() * 1000)),
"updatedAt": ev.get("at", int(time.time() * 1000)),
}
threads.append(new_th)
by_id[tid] = new_th
elif kind == "user_action":
tid = ev.get("threadId")
act = ev.get("action")
if tid in by_id:
th = by_id[tid]
if act == "accept":
th["status"] = "done"
th["promotedBy"] = "user_accepted"
elif act == "dismiss":
th["status"] = "dismissed"
elif act == "reopen":
th["status"] = "open"
elif act == "drop":
threads = [t for t in threads if t["id"] != tid]
del by_id[tid]
if tid in by_id:
by_id[tid]["updatedAt"] = ev.get("at", int(time.time() * 1000))
return threads
async def list(status: str | None = None) -> list[dict[str, Any]]:
"""List tracked commitment threads. Optional filter by status ('open', 'active', 'done', etc.)."""
all_threads = _read_threads()
if status is not None:
return [t for t in all_threads if t["status"] == status]
return all_threads
async def complete(thread_id: str, reason: str | None = None) -> dict[str, Any]:
"""Mark a thread complete (status: done)."""
if not isinstance(thread_id, str):
raise TypeError(f"thread_id must be str, got {type(thread_id).__name__}")
_append_event({
"kind": "user_action",
"action": "accept",
"threadId": thread_id,
"reason": reason,
"at": int(time.time() * 1000),
})
return {"status": "ok", "threadId": thread_id, "action": "accept"}
async def dismiss(thread_id: str, reason: str | None = None) -> dict[str, Any]:
"""Dismiss an obsolete or unwanted thread (status: dismissed)."""
if not isinstance(thread_id, str):
raise TypeError(f"thread_id must be str, got {type(thread_id).__name__}")
_append_event({
"kind": "user_action",
"action": "dismiss",
"threadId": thread_id,
"reason": reason,
"at": int(time.time() * 1000),
})
return {"status": "ok", "threadId": thread_id, "action": "dismiss"}
async def reopen(thread_id: str) -> dict[str, Any]:
"""Reopen a closed thread (status: open)."""
if not isinstance(thread_id, str):
raise TypeError(f"thread_id must be str, got {type(thread_id).__name__}")
_append_event({
"kind": "user_action",
"action": "reopen",
"threadId": thread_id,
"at": int(time.time() * 1000),
})
return {"status": "ok", "threadId": thread_id, "action": "reopen"}