#!/usr/bin/env python3 """Serve worker activity and keep its loading overlay scoped to resumed chats.""" from __future__ import annotations import argparse from pathlib import Path HELPER_MARKER = '''@app.get("/api/sessions/{session_id}/messages") async def get_session_messages(session_id: str, profile: Optional[str] = None): ''' HELPER_REPLACEMENT = '''def _run_activity_messages(session_id: str) -> List[Dict[str, Any]]: """Read a bounded activity journal without treating it as agent history.""" digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest() root = get_hermes_home() / "run-activity" if root.is_symlink(): return [] paths = ( root / f"{digest}.previous.jsonl", root / f"{digest}.jsonl", ) entries: List[Dict[str, Any]] = [] for path in paths: try: if path.is_symlink() or not path.is_file(): continue if path.stat().st_size > 600_000: continue for line in path.read_text(encoding="utf-8").splitlines(): item = json.loads(line) if not isinstance(item, dict): continue content = item.get("content") timestamp = item.get("timestamp") if not isinstance(content, str) or not isinstance( timestamp, (int, float) ): continue entries.append( { "role": "assistant", "content": content[:1_000], "timestamp": float(timestamp), "observed": True, } ) except (OSError, UnicodeError, ValueError, json.JSONDecodeError): continue return entries[-1_000:] def _present_activity_message(message: Dict[str, Any]) -> Dict[str, Any]: """Give a persisted tool-only assistant turn an honest visible summary. Hermes deliberately stores a blank assistant ``content`` field when the turn only invokes tools. The tool call records are retained separately, but a transcript card with neither prose nor a summary looks like missing history. This projection leaves the stored message intact and names only the already-recorded tool functions. """ item = dict(message) if item.get("role") != "assistant" or str(item.get("content") or "").strip(): return item calls = item.get("tool_calls") if not isinstance(calls, list): return item names: List[str] = [] for call in calls: function = call.get("function") if isinstance(call, dict) else None name = function.get("name") if isinstance(function, dict) else None if isinstance(name, str) and name.strip(): names.append(name.strip()[:80]) if names: item["content"] = "Calling " + ", ".join(names[:8]) + "." item["observed"] = True return item @app.get("/api/sessions/{session_id}/messages") async def get_session_messages( session_id: str, profile: Optional[str] = None, limit: Optional[int] = None, ): ''' MESSAGES_BEFORE = ''' messages = db.get_messages(sid) return {"session_id": sid, "messages": messages} ''' MESSAGES_AFTER = ''' messages = [ _present_activity_message(item) for item in [ *db.get_messages(sid), *_run_activity_messages(sid), ] ] messages.sort(key=lambda item: float(item.get("timestamp") or 0.0)) total_messages = len(messages) if limit is not None: messages = messages[-max(1, min(int(limit), 10_000)) :] return { "session_id": sid, "messages": messages, "total_messages": total_messages, } ''' LATEST_ROWS_BEFORE = ''' "SELECT id, parent_session_id, started_at FROM sessions" ).fetchall() for row in raw_rows: rows.append({ "id": row_get(row, "id", 0), "parent_session_id": row_get(row, "parent_session_id", 1), "started_at": row_get(row, "started_at", 2), }) ''' LATEST_ROWS_AFTER = ''' "SELECT id, parent_session_id, started_at, ended_at FROM sessions" ).fetchall() for row in raw_rows: rows.append({ "id": row_get(row, "id", 0), "parent_session_id": row_get(row, "parent_session_id", 1), "started_at": row_get(row, "started_at", 2), "ended_at": row_get(row, "ended_at", 3), }) ''' LATEST_SELECTION_BEFORE = ''' children = {} for row in rows: rid = row.get("id") parent = row.get("parent_session_id") if rid and parent: children.setdefault(parent, []).append(row) def started(row): try: return float(row.get("started_at") or 0) except Exception: return 0.0 current = sid path = [sid] seen = {sid} while children.get(current): candidates = [r for r in children[current] if r.get("id") not in seen] if not candidates: break candidates.sort(key=started, reverse=True) current = candidates[0]["id"] path.append(current) seen.add(current) return current, path ''' LATEST_SELECTION_AFTER = ''' children = {} rows_by_id = {} for row in rows: rid = row.get("id") parent = row.get("parent_session_id") if rid: rows_by_id[rid] = row if rid and parent: children.setdefault(parent, []).append(row) def started(row): try: return float(row.get("started_at") or 0) except Exception: return 0.0 # Old dashboard versions rewrote the URL to a delegated leaf without # retaining its objective. Recover only the immediate objective parent. # Climbing to the oldest ancestor mixes unrelated workstreams that happen # to share a long-lived dashboard/handoff session. anchor = sid if not children.get(anchor): parent = (rows_by_id.get(anchor) or {}).get("parent_session_id") if parent and parent in rows_by_id: anchor = parent sid = anchor root = rows_by_id.get(sid) or db.get_session(sid) or {"id": sid} descendants = [(root, [sid])] stack = [(sid, [sid])] seen = {sid} while stack: parent, parent_path = stack.pop() candidates = sorted(children.get(parent, []), key=started, reverse=True) for row in candidates: child = row.get("id") if not child or child in seen: continue seen.add(child) child_path = [*parent_path, child] descendants.append((row, child_path)) stack.append((child, child_path)) # A durable parent can resume after a delegated reviewer exits. Prefer the # newest still-open member of the lineage instead of stranding the browser # on the most recently created (but already finished) child. # An ended objective can retain orphaned children whose process died before # SessionDB recorded ended_at. Do not let those stale rows defeat a newer, # completed repair/review branch. While the objective itself is live, an # open child or the resumed parent still wins as before. active = ( [item for item in descendants if item[0].get("ended_at") is None] if root.get("ended_at") is None else [] ) if active: row, path = max(active, key=lambda item: (started(item[0]), len(item[1]))) return row.get("id") or sid, path leaves = [ item for item in descendants if not children.get(item[0].get("id")) ] row, path = max(leaves or descendants, key=lambda item: started(item[0])) return row.get("id") or sid, path ''' SPA_INDEX_BEFORE = ''' html = _index_path.read_text(encoding="utf-8") ''' SPA_INDEX_AFTER = SPA_INDEX_BEFORE + ''' # The bundled overlay's inline display:flex overrides HTML hidden. # Honor hidden before React mounts, including ordinary dashboard URLs. html = html.replace( "", "", 1, ) ''' def patch(source: Path, destination: Path) -> None: """Apply the activity projection and fail closed on upstream drift.""" content = source.read_text(encoding="utf-8") if HELPER_MARKER not in content: raise RuntimeError("Hermes dashboard activity helper context changed") if MESSAGES_BEFORE not in content: raise RuntimeError("Hermes dashboard messages context changed") if LATEST_ROWS_BEFORE not in content: raise RuntimeError("Hermes dashboard lineage row context changed") if LATEST_SELECTION_BEFORE not in content: raise RuntimeError("Hermes dashboard lineage selection context changed") if content.count(SPA_INDEX_BEFORE) != 1: raise RuntimeError("Hermes dashboard bootstrap context changed") content = content.replace(HELPER_MARKER, HELPER_REPLACEMENT, 1) content = content.replace(LATEST_ROWS_BEFORE, LATEST_ROWS_AFTER, 1) content = content.replace(LATEST_SELECTION_BEFORE, LATEST_SELECTION_AFTER, 1) content = content.replace(SPA_INDEX_BEFORE, SPA_INDEX_AFTER, 1) destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text( content.replace(MESSAGES_BEFORE, MESSAGES_AFTER, 1), encoding="utf-8", ) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("source", type=Path) parser.add_argument("destination", type=Path) args = parser.parse_args() patch(args.source, args.destination) return 0 if __name__ == "__main__": raise SystemExit(main())