From f031c479ba4a769608421709da25ba1c2903c885 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 12 Jul 2026 00:45:18 -0500 Subject: [PATCH 01/12] refac --- cptr/frontend/src/lib/utils/diff.ts | 57 +++++++++++++---------------- 1 file changed, 26 insertions(+), 31 deletions(-) diff --git a/cptr/frontend/src/lib/utils/diff.ts b/cptr/frontend/src/lib/utils/diff.ts index 7737b056..81fca197 100644 --- a/cptr/frontend/src/lib/utils/diff.ts +++ b/cptr/frontend/src/lib/utils/diff.ts @@ -9,6 +9,9 @@ export type DiffLineGroup = { type: T['type']; lines export type SplitDiffRow = { oldLine: InlineDiffLine | null; newLine: InlineDiffLine | null }; export type DiffStats = { additions: number; deletions: number }; +// GitHub Desktop uses the same cutoff for intraline highlighting. +const MAX_INTRA_LINE_DIFF_STRING_LENGTH = 1024; + export function countDiffStats(files: DiffFile[]): DiffStats { let additions = 0; let deletions = 0; @@ -105,11 +108,16 @@ export function withInlineDiffSegments(lines: NumberedDiffLine[]): InlineDiffLin const removed = block.filter((item) => item.type === 'removed'); const added = block.filter((item) => item.type === 'added'); + const canHighlightChanges = removed.length === added.length; const removedSegments = removed.map((item, index) => - segmentsForPair(item.content, added[index]?.content, 'removed') + canHighlightChanges + ? segmentsForPair(item.content, added[index].content, 'removed') + : unchangedSegments(item.content) ); const addedSegments = added.map((item, index) => - segmentsForPair(removed[index]?.content, item.content, 'added') + canHighlightChanges + ? segmentsForPair(removed[index].content, item.content, 'added') + : unchangedSegments(item.content) ); for (const item of block) { @@ -162,15 +170,17 @@ export function splitDiffRows(lines: InlineDiffLine[]): SplitDiffRow[] { } function segmentsForPair( - oldText: string | undefined, - newText: string | undefined, + oldText: string, + newText: string, side: 'removed' | 'added' ): InlineDiffSegment[] { const text = side === 'removed' ? oldText : newText; - if (text === undefined) return []; - if (oldText === undefined || newText === undefined) - return [{ text: text || ' ', changed: false }]; - if (oldText === newText) return [{ text: text || ' ', changed: false }]; + if ( + oldText === newText || + oldText.length >= MAX_INTRA_LINE_DIFF_STRING_LENGTH || + newText.length >= MAX_INTRA_LINE_DIFF_STRING_LENGTH + ) + return unchangedSegments(text); let prefix = 0; const maxPrefix = Math.min(oldText.length, newText.length); @@ -187,31 +197,16 @@ function segmentsForPair( newSuffix -= 1; } - const oldChangedLength = oldSuffix - prefix; - const newChangedLength = newSuffix - prefix; - const sharedLength = Math.max( - oldText.length - oldChangedLength, - newText.length - newChangedLength - ); - const similarity = sharedLength / Math.max(oldText.length, newText.length, 1); - const changedLength = side === 'removed' ? oldChangedLength : newChangedLength; - if (similarity < 0.55 || changedLength / Math.max(text.length, 1) > 0.45) { - return [{ text: text || ' ', changed: false }]; - } - const end = side === 'removed' ? oldSuffix : newSuffix; - const changed = text.slice(prefix, end); - const tokenStart = changed.search(/[A-Za-z0-9_$]/); - if (tokenStart < 0) return [{ text: text || ' ', changed: false }]; - - const tokenEnd = changed.search(/[A-Za-z0-9_$][^A-Za-z0-9_$]*$/); - const highlightStart = tokenStart >= 0 ? prefix + tokenStart : prefix; - const highlightEnd = tokenEnd >= 0 ? prefix + tokenEnd + 1 : end; const segments = [ - { text: text.slice(0, highlightStart), changed: false }, - { text: text.slice(highlightStart, highlightEnd), changed: true }, - { text: text.slice(highlightEnd), changed: false } + { text: text.slice(0, prefix), changed: false }, + { text: text.slice(prefix, end), changed: true }, + { text: text.slice(end), changed: false } ].filter((segment) => segment.text.length > 0); - return segments.length ? segments : [{ text: ' ', changed: false }]; + return segments.length ? segments : unchangedSegments(text); +} + +function unchangedSegments(text: string): InlineDiffSegment[] { + return [{ text: text || ' ', changed: false }]; } From e7afb6cba3f1c0029328ed1a34894f6acded5bcd Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 12 Jul 2026 00:57:58 -0500 Subject: [PATCH 02/12] refac --- cptr/app.py | 11 ++ cptr/env.py | 1 + cptr/events.py | 21 ++- cptr/models/__init__.py | 4 +- cptr/models/chats.py | 89 ++++++++++- cptr/routers/chat.py | 22 ++- cptr/socket/main.py | 9 ++ cptr/utils/async_subagents.py | 8 + cptr/utils/chat_task.py | 4 +- cptr/utils/timers.py | 287 ++++++++++++++++++++++++++++++++++ cptr/utils/tools.py | 125 ++++++++++++--- 11 files changed, 550 insertions(+), 31 deletions(-) create mode 100644 cptr/utils/timers.py diff --git a/cptr/app.py b/cptr/app.py index a0421af6..16e3883d 100644 --- a/cptr/app.py +++ b/cptr/app.py @@ -71,6 +71,11 @@ async def lifespan(app: FastAPI): app.state.scheduler_task = asyncio.create_task(scheduler_worker_loop(app)) + from cptr.utils.timers import recover_timers, timer_worker_loop + + await recover_timers() + app.state.timer_task = asyncio.create_task(timer_worker_loop(app)) + # Start messaging bots from cptr.utils.bridge import BotManager @@ -80,6 +85,12 @@ async def lifespan(app: FastAPI): try: yield finally: + timer_task = getattr(app.state, "timer_task", None) + if timer_task: + timer_task.cancel() + with suppress(asyncio.CancelledError): + await timer_task + scheduler_task = getattr(app.state, "scheduler_task", None) if scheduler_task: scheduler_task.cancel() diff --git a/cptr/env.py b/cptr/env.py index fd9f47aa..4cc33351 100644 --- a/cptr/env.py +++ b/cptr/env.py @@ -82,6 +82,7 @@ def _env_int(name: str, default: int) -> int: # ── Automation scheduler ──────────────────────────────────── AUTOMATION_POLL_INTERVAL = int(os.environ.get("AUTOMATION_POLL_INTERVAL", "10")) +TIMER_POLL_INTERVAL = int(os.environ.get("TIMER_POLL_INTERVAL", "10")) # ── CORS ──────────────────────────────────────────────────── # Socket.IO CORS allowed origins. diff --git a/cptr/events.py b/cptr/events.py index d871b956..a2113710 100644 --- a/cptr/events.py +++ b/cptr/events.py @@ -26,6 +26,16 @@ def label(self) -> str: class EventDefinitions: + CHAT_READ = EventDefinition( + "chat.read", + "The user explicitly marked a chat as read.", + "Chat read", + ) + CHAT_USER_MESSAGE = EventDefinition( + "chat.user_message", + "A user message was added to a chat.", + "User message", + ) CHAT_FINISHED = EventDefinition( "chat.finished", "A chat run finished successfully.", @@ -191,7 +201,16 @@ async def handle_event(self, event: Event) -> None: await dispatch_notification_event(event) -EVENT_SINKS = [NotificationEventSink()] +class TimerEventSink: + async def handle_event(self, event: Event) -> None: + if event.event not in {EVENTS.CHAT_READ.name, EVENTS.CHAT_USER_MESSAGE.name}: + return + from cptr.utils.timers import cancel_timers_for_event + + await cancel_timers_for_event(event) + + +EVENT_SINKS = [TimerEventSink(), NotificationEventSink()] async def publish_event( diff --git a/cptr/models/__init__.py b/cptr/models/__init__.py index 3727577a..b6a0a850 100644 --- a/cptr/models/__init__.py +++ b/cptr/models/__init__.py @@ -5,7 +5,7 @@ from cptr.models.workspaces import Workspace from cptr.models.config import Config from cptr.models.files import File -from cptr.models.chats import Chat, ChatMessage +from cptr.models.chats import Chat, ChatMessage, is_internal_chat from cptr.models.automations import Automation, AutomationRun __all__ = [ @@ -18,7 +18,7 @@ "File", "Chat", "ChatMessage", + "is_internal_chat", "Automation", "AutomationRun", ] - diff --git a/cptr/models/chats.py b/cptr/models/chats.py index a59266a1..a954cd47 100644 --- a/cptr/models/chats.py +++ b/cptr/models/chats.py @@ -27,6 +27,12 @@ def _uuid() -> str: return str(uuid.uuid4()) +def is_internal_chat(meta: dict | None) -> bool: + """Whether a chat is harness-owned and should stay out of normal chat surfaces.""" + meta = meta or {} + return bool(meta.get("internal") or meta.get("subagent")) + + class Chat(Base): """A chat conversation. Workspace association lives in the filesystem.""" @@ -59,6 +65,71 @@ async def get_by_ids(chat_ids: list[str]) -> list[Chat]: result = await db.execute(select(Chat).where(Chat.id.in_(chat_ids))) return list(result.scalars().all()) + @staticmethod + async def get_due_timers(now_ns: int, limit: int = 10) -> list[Chat]: + """Return pending internal timer chats whose due time has arrived.""" + async with await get_db() as db: + result = await db.execute(select(Chat).where(Chat.meta.is_not(None))) + timers = [ + chat + for chat in result.scalars().all() + if (chat.meta or {}).get("internal") is True + and (chat.meta or {}).get("type") == "timer" + and (chat.meta or {}).get("timer_status") == "pending" + and int((chat.meta or {}).get("timer_at") or 0) <= now_ns + ] + timers.sort(key=lambda chat: int((chat.meta or {}).get("timer_at") or 0)) + return timers[:limit] + + @staticmethod + async def get_pending_timers(parent_chat_id: str) -> list[Chat]: + """Return pending timer children for a parent chat.""" + async with await get_db() as db: + result = await db.execute(select(Chat).where(Chat.meta.is_not(None))) + return [ + chat + for chat in result.scalars().all() + if (chat.meta or {}).get("internal") is True + and (chat.meta or {}).get("type") == "timer" + and (chat.meta or {}).get("parent_chat_id") == parent_chat_id + and (chat.meta or {}).get("timer_status") == "pending" + ] + + @staticmethod + async def get_timers(status: str | None = None) -> list[Chat]: + """Return internal timer chats, optionally in one lifecycle state.""" + async with await get_db() as db: + result = await db.execute(select(Chat).where(Chat.meta.is_not(None))) + return [ + chat + for chat in result.scalars().all() + if (chat.meta or {}).get("internal") is True + and (chat.meta or {}).get("type") == "timer" + and (status is None or (chat.meta or {}).get("timer_status") == status) + ] + + @staticmethod + async def get_internal_descendants(parent_chat_id: str) -> list[Chat]: + """Return every internal child below a chat, including nested timers.""" + async with await get_db() as db: + result = await db.execute(select(Chat).where(Chat.meta.is_not(None))) + candidates = list(result.scalars().all()) + children: list[Chat] = [] + parent_ids = {parent_chat_id} + while parent_ids: + found = [ + chat + for chat in candidates + if is_internal_chat(chat.meta) + and (chat.meta or {}).get("parent_chat_id") in parent_ids + ] + if not found: + break + children.extend(found) + parent_ids = {chat.id for chat in found} + candidates = [chat for chat in candidates if chat not in found] + return children + @staticmethod async def create( user_id: str, @@ -173,9 +244,21 @@ async def update_last_read_at(chat_id: str, user_id: str, last_read_at: int) -> @staticmethod async def delete(chat_id: str) -> bool: async with await get_db() as db: + result = await db.execute(select(Chat).where(Chat.meta.is_not(None))) + candidates = list(result.scalars().all()) + delete_ids = {chat_id} + frontier = {chat_id} + while frontier: + frontier = { + chat.id + for chat in candidates + if is_internal_chat(chat.meta) + and (chat.meta or {}).get("parent_chat_id") in frontier + } + delete_ids.update(frontier) # Messages cascade via FK, but be explicit - await db.execute(delete(ChatMessage).where(ChatMessage.chat_id == chat_id)) - result = await db.execute(delete(Chat).where(Chat.id == chat_id)) + await db.execute(delete(ChatMessage).where(ChatMessage.chat_id.in_(delete_ids))) + result = await db.execute(delete(Chat).where(Chat.id.in_(delete_ids))) await db.commit() return result.rowcount > 0 @@ -237,7 +320,7 @@ async def search_by_text( def allowed(chat: Chat) -> bool: meta = chat.meta or {} - if not include_subagents and meta.get("subagent"): + if not include_subagents and is_internal_chat(meta): return False if workspace and meta.get("workspace") != workspace: return False diff --git a/cptr/routers/chat.py b/cptr/routers/chat.py index f671210d..d3364374 100644 --- a/cptr/routers/chat.py +++ b/cptr/routers/chat.py @@ -11,7 +11,7 @@ from fastapi import APIRouter, HTTPException, Query, Request from pydantic import BaseModel -from cptr.models import Chat, ChatMessage, Config +from cptr.models import Chat, ChatMessage, Config, is_internal_chat from cptr.utils.config import check_access, now_ms, _get_jwt_secret from cptr.utils.crypto import decrypt_key from cptr.utils.workspace import ensure_cptr_gitignored @@ -97,6 +97,8 @@ def _scan_chat_files() -> list[dict]: continue chat_meta = chat.meta if isinstance(chat.meta, dict) else {} + if is_internal_chat(chat_meta): + continue if chat.meta is not None and not isinstance(chat.meta, dict): log.warning( "chat.list.bad_meta workspace=%r chat_id=%s title=%r meta_type=%s", @@ -581,10 +583,14 @@ async def delete_chat(chat_id: str, request: Request): if not chat or chat.user_id != user_id: raise HTTPException(404, "chat not found") - # Remove the workspace chat file. + # Remove the workspace chat file and durable internal children. workspace = chat.meta.get("workspace") if chat.meta else None chat_file = chat_directory(workspace) / f"{chat_id}.json" await asyncio.to_thread(chat_file.unlink, True) # missing_ok=True + for child in await Chat.get_internal_descendants(chat_id): + child_workspace = (child.meta or {}).get("workspace") + child_file = chat_directory(child_workspace) / f"{child.id}.json" + await asyncio.to_thread(child_file.unlink, True) await Chat.delete(chat_id) from cptr.socket.main import emit_to_user @@ -801,6 +807,18 @@ async def send_message(body: SendMessageRequest, request: Request): # Update chat pointer to new leaf while still holding the input lock. await Chat.update_current_message(chat.id, assistant_msg.id, now_ms()) + submitted_message = queued_msg or user_msg + if submitted_message: + from cptr.events import EVENTS, publish_event + + await publish_event( + EVENTS.CHAT_USER_MESSAGE, + actor={"id": user_id}, + subject_id=chat.id, + subject_type="chat", + source="chat", + ) + if queued_msg: await process_pending_chat_inputs(chat.id, user_id, workspace or "") return {"chat_id": chat.id, "message_id": queued_msg.id, "queued": True} diff --git a/cptr/socket/main.py b/cptr/socket/main.py index d0c27876..c9ca6b7c 100644 --- a/cptr/socket/main.py +++ b/cptr/socket/main.py @@ -89,6 +89,15 @@ async def on_chat_read(sid, data): last_read_at = now_ms() if await Chat.update_last_read_at(chat_id, user_id, last_read_at): + from cptr.events import EVENTS, publish_event + + await publish_event( + EVENTS.CHAT_READ, + actor={"id": user_id}, + subject_id=chat_id, + subject_type="chat", + source="socket", + ) chat = await Chat.get_by_id(chat_id) workspace = (chat.meta or {}).get("workspace", "") if chat else "" from cptr.utils.chat_task import get_active_chat_ids diff --git a/cptr/utils/async_subagents.py b/cptr/utils/async_subagents.py index f1eb0f47..c1988e26 100644 --- a/cptr/utils/async_subagents.py +++ b/cptr/utils/async_subagents.py @@ -160,6 +160,14 @@ async def _finalize( snapshot = {k: v for k, v in record.items() if k != "task"} _prune_completed_locked() + if snapshot.get("timer_chat_id"): + from cptr.utils.timers import set_timer_completion + + try: + await set_timer_completion(snapshot, status, error) + except Exception: + logger.exception("Failed to record timer completion %s", delegation_id) + injector = _completion_injector_override or _inject_completion try: await injector(snapshot) diff --git a/cptr/utils/chat_task.py b/cptr/utils/chat_task.py index 738722c2..ece91220 100644 --- a/cptr/utils/chat_task.py +++ b/cptr/utils/chat_task.py @@ -23,7 +23,7 @@ load_skill, ) from cptr.utils.summarize import summarize_messages -from cptr.models import Chat, ChatMessage, Config +from cptr.models import Chat, ChatMessage, Config, is_internal_chat from cptr.socket.main import emit_to_user from cptr.utils.ai import ( ChatCompletionForm, @@ -1936,7 +1936,7 @@ async def _finish_reasoning_item(): tools = [t for t in tools if t["name"] != "view_skill"] # Strip delegate_task from sub-agent chats (depth limit = 1) - if chat_obj and (chat_obj.meta or {}).get("subagent"): + if chat_obj and is_internal_chat(chat_obj.meta): tools = [t for t in tools if t["name"] not in {"delegate_task", "update_memory"}] # Parse $skill-name mentions from the user message to auto-activate skills diff --git a/cptr/utils/timers.py b/cptr/utils/timers.py new file mode 100644 index 00000000..6fc3183b --- /dev/null +++ b/cptr/utils/timers.py @@ -0,0 +1,287 @@ +"""Durable one-shot timers backed by dormant internal child chats.""" + +from __future__ import annotations + +import asyncio +import logging +import re +import time +from datetime import datetime + +from cptr.env import TIMER_POLL_INTERVAL + +logger = logging.getLogger(__name__) + +_RELATIVE_TIME = re.compile(r"^\+(\d+)([smhd])$") +_RFC3339_TIME = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$") +_TIME_UNITS_NS = { + "s": 1_000_000_000, + "m": 60 * 1_000_000_000, + "h": 60 * 60 * 1_000_000_000, + "d": 24 * 60 * 60 * 1_000_000_000, +} +_manager_lock = asyncio.Lock() + + +def parse_timer_at(value: str) -> int: + """Normalize a relative offset or timezone-aware RFC 3339 timestamp.""" + raw = value.strip() + now = time.time_ns() + relative = _RELATIVE_TIME.fullmatch(raw) + if relative: + count = int(relative.group(1)) + if count <= 0: + raise ValueError("at must be in the future.") + return now + count * _TIME_UNITS_NS[relative.group(2)] + + if not _RFC3339_TIME.fullmatch(raw): + raise ValueError("at must be +5s, +5m, +1h, +2d, or an RFC 3339 timestamp with a timezone.") + try: + parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError( + "at must be +5s, +5m, +1h, +2d, or an RFC 3339 timestamp with a timezone." + ) from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise ValueError("absolute at values must include an explicit timezone.") + + due_at = int(parsed.timestamp() * 1_000_000_000) + if due_at <= now: + raise ValueError("at must be in the future.") + return due_at + + +async def cancel_timers_for_event(event) -> None: + """Cancel matching dormant children after a committed parent-chat event.""" + subject = event.subject or {} + if subject.get("type") != "chat" or not subject.get("id"): + return + + from cptr.models import Chat + from cptr.utils.config import now_ms + + async with _manager_lock: + timers = await Chat.get_pending_timers(str(subject["id"])) + for timer in timers: + meta = dict(timer.meta or {}) + if event.event not in (meta.get("cancel_on") or []): + continue + meta.update( + { + "timer_status": "cancelled", + "timer_cancelled_at": time.time_ns(), + "timer_cancelled_by": event.event, + } + ) + await Chat.update_meta(timer.id, meta, now_ms()) + + +def _pack_parent_context(parent, messages, checkpoint_id: str | None) -> str: + """Build a small launch-time view of the parent without changing its history.""" + message_by_id = {message.id: message for message in messages} + branch = [] + current = message_by_id.get(parent.current_message_id) + while current: + branch.append(current) + current = message_by_id.get(current.parent_id) + branch.reverse() + + checkpoint_index = next( + (index for index, message in enumerate(branch) if message.id == checkpoint_id), -1 + ) + changes = branch[checkpoint_index + 1 :] if checkpoint_index >= 0 else branch[-12:] + changes = changes[-12:] + + parts = [] + summary = next( + (message.chat_summary for message in reversed(branch) if message.chat_summary), None + ) + if summary: + parts.append(f"## Parent summary\n{summary[-4_000:]}") + if changes: + lines = [] + remaining = 6_000 + for message in changes: + content = (message.content or "").strip() + if not content: + continue + content = content[: min(len(content), remaining)] + lines.append(f"{message.role}: {content}") + remaining -= len(content) + if remaining <= 0: + break + if lines: + parts.append("## Changes since timer was set\n" + "\n\n".join(lines)) + return "\n\n".join(parts) + + +async def _set_timer_status(chat_id: str, status: str, **fields) -> None: + from cptr.models import Chat + from cptr.utils.config import now_ms + + chat = await Chat.get_by_id(chat_id) + if not chat: + return + meta = dict(chat.meta or {}) + meta["timer_status"] = status + meta.update(fields) + await Chat.update_meta(chat_id, meta, now_ms()) + + +async def set_timer_completion(record: dict, status: str, error: str | None) -> None: + """Persist completion before the normal async-subagent result reaches the parent.""" + timer_chat_id = record.get("timer_chat_id") + if not timer_chat_id: + return + final_status = "completed" if status == "completed" else "error" + await _set_timer_status( + timer_chat_id, + final_status, + timer_completed_at=time.time_ns(), + timer_error=error, + ) + + +async def _launch_timer(timer, app) -> None: + from cptr.models import Chat, ChatMessage + from cptr.utils.async_subagents import ( + attach_subagent_chat, + reserve_async_subagent, + start_async_subagent, + ) + from cptr.utils.chat_export import export_chat_to_file + from cptr.utils.config import now_ms + from cptr.utils.model_targets import ApiModelTarget, resolve_model_target + from cptr.utils.tools import _get_subagent_config, _run_existing_subagent_chat + + async with _manager_lock: + timer = await Chat.get_by_id(timer.id) + if not timer: + return + meta = dict(timer.meta or {}) + if meta.get("timer_status") != "pending" or int(meta.get("timer_at") or 0) > time.time_ns(): + return + + parent = await Chat.get_by_id(meta.get("parent_chat_id", "")) + if not parent: + await _set_timer_status(timer.id, "error", timer_error="parent chat no longer exists") + return + + config = await _get_subagent_config() + if not config["background_enabled"]: + return + + try: + target = await resolve_model_target(meta["timer_model_id"], app.state) + except Exception as exc: # model configuration can change while a timer waits + await _set_timer_status(timer.id, "error", timer_error=f"model unavailable: {exc}") + return + if not isinstance(target, ApiModelTarget): + await _set_timer_status( + timer.id, "error", timer_error="background timers require an API model" + ) + return + + task_message = await ChatMessage.get_by_id(timer.current_message_id) + if not task_message: + await _set_timer_status(timer.id, "error", timer_error="timer task message is missing") + return + + reserve = await reserve_async_subagent( + config["max_async"], + task=task_message.content, + context="", + workspace=meta.get("workspace", ""), + user_id=timer.user_id, + parent_chat_id=parent.id, + parent_message_id=meta.get("timer_parent_message_id"), + connection=target.connection, + model=target.runtime_model, + model_id=target.full_model_id, + timer_chat_id=timer.id, + ) + if reserve.get("status") == "rejected": + return + + parent_messages = await ChatMessage.get_all_by_chat(parent.id) + parent_context = _pack_parent_context( + parent, parent_messages, meta.get("timer_parent_message_id") + ) + content = task_message.content + if parent_context: + content = f"{content}\n\n## Parent context at launch\n{parent_context}" + await ChatMessage.update(task_message.id, content=content) + + assistant_msg = await ChatMessage.create( + chat_id=timer.id, + role="assistant", + content="", + parent_id=task_message.id, + model=target.full_model_id, + done=False, + created_at=now_ms(), + ) + meta.update( + { + "timer_status": "running", + "timer_started_at": time.time_ns(), + "timer_assistant_message_id": assistant_msg.id, + "delegation_id": reserve["delegation_id"], + } + ) + await Chat.update_meta(timer.id, meta, now_ms()) + await Chat.update_current_message(timer.id, assistant_msg.id, now_ms()) + await export_chat_to_file(timer.id) + await attach_subagent_chat( + reserve["delegation_id"], + subagent_chat_id=timer.id, + subagent_message_id=assistant_msg.id, + ) + + async def runner() -> str: + return await _run_existing_subagent_chat( + assistant_msg_id=assistant_msg.id, + chat_id=timer.id, + workspace=meta.get("workspace", ""), + connection=target.connection, + model=target.runtime_model, + user_id=timer.user_id, + config=config, + ) + + await start_async_subagent(reserve["delegation_id"], runner) + + +async def timer_worker_loop(app) -> None: + """Poll durable pending timers; a capacity miss simply waits for the next pass.""" + from cptr.models import Chat + + logger.info("Timer worker started (poll interval: %ds)", TIMER_POLL_INTERVAL) + while True: + try: + due = await Chat.get_due_timers(time.time_ns()) + for timer in due: + await _launch_timer(timer, app) + except Exception: + logger.exception("Timer worker error") + await asyncio.sleep(TIMER_POLL_INTERVAL) + + +async def recover_timers() -> None: + """Do not replay work that was already launched before a process restart.""" + from cptr.models import Chat, ChatMessage + from cptr.utils.config import now_ms + + for timer in await Chat.get_timers("pending"): + current = await ChatMessage.get_by_id(timer.current_message_id) + if current and current.role == "assistant": + await ChatMessage.delete(current.id) + await Chat.update_current_message(timer.id, current.parent_id, now_ms()) + + for timer in await Chat.get_timers("running"): + await _set_timer_status( + timer.id, + "error", + timer_completed_at=time.time_ns(), + timer_error="interrupted by restart", + ) diff --git a/cptr/utils/tools.py b/cptr/utils/tools.py index 076d3371..bd8e11f8 100644 --- a/cptr/utils/tools.py +++ b/cptr/utils/tools.py @@ -1997,7 +1997,7 @@ async def search_chats( """ from sqlalchemy import select - from cptr.models import Chat, ChatMessage + from cptr.models import Chat, ChatMessage, is_internal_chat from cptr.utils.db import get_db user_id = __context__.get("user_id") @@ -2022,8 +2022,8 @@ async def get_allowed_chat(cid: str): if not chat or chat.user_id != user_id: return None, "chat not found" meta = chat.meta or {} - if not include_subagents and meta.get("subagent"): - return None, "chat is a sub-agent chat" + if not include_subagents and is_internal_chat(meta): + return None, "chat is an internal chat" if workspace and meta.get("workspace") != workspace: return None, "chat is outside the current workspace" return chat, None @@ -2124,7 +2124,7 @@ async def get_allowed_chat(cid: str): meta = chat.meta or {} if chat.id == current_chat_id: continue - if not include_subagents and meta.get("subagent"): + if not include_subagents and is_internal_chat(meta): continue if workspace and meta.get("workspace") != workspace: continue @@ -2299,7 +2299,7 @@ async def delegate_task( delegation_id = reserve["delegation_id"] try: - chat, assistant_msg = await _create_subagent_chat( + chat, _, assistant_msg = await _create_subagent_chat( task=task, context=context, workspace=__context__["workspace"], @@ -2369,6 +2369,76 @@ async def _runner() -> str: ) +async def timer( + prompt: str, + at: str, + cancel_on: list[Literal["chat.read", "chat.user_message"]] | None = None, + *, + __context__: dict, +) -> str: + """Set a one-shot timer for this chat. + + Use it when time is the missing input: wait for a deployment, retry after + backoff, revisit work after a deadline, or follow up after silence. At + `at`, the agent receives `prompt` and the latest thread context; it may + act, set another timer, or finish silently. + + `cancel_on` prevents launch when `chat.read` or `chat.user_message` + happens first in this chat. Omit it when timed work must run regardless. + """ + from cptr.utils.timers import parse_timer_at + + if not prompt.strip(): + return "Error: prompt must not be empty." + + try: + due_at = parse_timer_at(at) + except ValueError as exc: + return f"Error: {exc}" + + selected_events = cancel_on or [] + allowed_events = {"chat.read", "chat.user_message"} + if any(event not in allowed_events for event in selected_events): + return "Error: cancel_on accepts only chat.read and chat.user_message." + selected_events = list(dict.fromkeys(selected_events)) + + config = await _get_subagent_config() + if not config["background_enabled"]: + return "Error: background sub-agents are disabled in settings." + + full_model_id = __context__.get("full_model_id") or __context__["model_id"] + chat, _, _ = await _create_subagent_chat( + task=prompt, + context="", + workspace=__context__["workspace"], + model=full_model_id, + user_id=__context__["user_id"], + parent_chat_id=__context__["chat_id"], + config=config, + child_type="timer", + deferred=True, + extra_meta={ + "timer_at": due_at, + "cancel_on": selected_events, + "timer_status": "pending", + "timer_parent_message_id": __context__.get("message_id"), + "timer_model_id": full_model_id, + }, + ) + + from datetime import datetime, timezone + + return json.dumps( + { + "status": "set", + "at": datetime.fromtimestamp(due_at / 1_000_000_000, timezone.utc) + .isoformat() + .replace("+00:00", "Z"), + "cancel_on": selected_events, + } + ) + + async def _create_subagent_chat( task: str, context: str, @@ -2378,6 +2448,10 @@ async def _create_subagent_chat( parent_chat_id: str, config: dict, delegation_id: str | None = None, + *, + child_type: str = "subagent", + deferred: bool = False, + extra_meta: dict | None = None, ): """Create the real chat/messages used by a sub-agent.""" from cptr.models import Chat, ChatMessage @@ -2387,7 +2461,8 @@ async def _create_subagent_chat( user_content = f"{task}\n\n## Context\n{context}" if context else task meta = { "workspace": workspace, - "subagent": True, + "internal": True, + "type": child_type, "parent_chat_id": parent_chat_id, "params": { "tool_approval_mode": "full", # auto-approve all tools @@ -2395,10 +2470,12 @@ async def _create_subagent_chat( } if delegation_id: meta["delegation_id"] = delegation_id + if extra_meta: + meta.update(extra_meta) chat = await Chat.create( user_id=user_id, - title=f"Sub-agent: {task[:60]}", + title=f"{child_type.title()}: {task[:60]}", meta=meta, created_at=now_ms(), ) @@ -2410,19 +2487,22 @@ async def _create_subagent_chat( created_at=now_ms(), ) - assistant_msg = await ChatMessage.create( - chat_id=chat.id, - role="assistant", - content="", - parent_id=user_msg.id, - model=model, - done=False, - created_at=now_ms(), - ) - - await Chat.update_current_message(chat.id, assistant_msg.id, now_ms()) + assistant_msg = None + if deferred: + await Chat.update_current_message(chat.id, user_msg.id, now_ms()) + else: + assistant_msg = await ChatMessage.create( + chat_id=chat.id, + role="assistant", + content="", + parent_id=user_msg.id, + model=model, + done=False, + created_at=now_ms(), + ) + await Chat.update_current_message(chat.id, assistant_msg.id, now_ms()) await export_chat_to_file(chat.id) - return chat, assistant_msg + return chat, user_msg, assistant_msg async def _run_existing_subagent_chat( @@ -2469,7 +2549,7 @@ async def _run_subagent_chat( config: dict, ) -> str: """Create a real chat and run the agent loop on it.""" - chat, assistant_msg = await _create_subagent_chat( + chat, _, assistant_msg = await _create_subagent_chat( task=task, context=context, workspace=workspace, @@ -2491,6 +2571,7 @@ async def _run_subagent_chat( SUBAGENT_TOOLS: dict[str, dict] = { "delegate_task": {"fn": delegate_task, "auto": True}, + "timer": {"fn": timer, "auto": False}, } # Combined lookup for execution and approval (always available regardless of config) @@ -2811,7 +2892,7 @@ def _parse_param_descriptions(docstring: str) -> dict[str, str]: def _fn_to_schema(name: str, fn) -> dict: """Introspect function → {name, description, parameters} for LLM.""" doc = inspect.getdoc(fn) or "" - description = doc.split("\n")[0] + description = doc if name == "timer" else doc.split("\n")[0] param_descs = _parse_param_descriptions(doc) hints = get_type_hints(fn) sig = inspect.signature(fn) @@ -2895,6 +2976,8 @@ async def get_tool_list(builtin_tools: dict | None = None, workspace: str = "") "true", "1", ) + if not background_subagents_enabled: + tools.pop("timer", None) images_generation_enabled = (await Config.get("images.generation_enabled")) in ( True, "true", From e68ece23ea397f8bc9555f1a5d70abcf3a276b6d Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 12 Jul 2026 01:02:49 -0500 Subject: [PATCH 03/12] refac --- .../src/lib/components/chat/ChatInput.svelte | 5 ++++- .../src/lib/components/chat/ChatPanel.svelte | 18 ++++++++++++++++-- .../src/lib/components/chat/PlusMenu.svelte | 7 +++++-- cptr/frontend/src/lib/stores.ts | 17 +++++++++++++++++ 4 files changed, 42 insertions(+), 5 deletions(-) diff --git a/cptr/frontend/src/lib/components/chat/ChatInput.svelte b/cptr/frontend/src/lib/components/chat/ChatInput.svelte index 8206d2bc..884cb828 100644 --- a/cptr/frontend/src/lib/components/chat/ChatInput.svelte +++ b/cptr/frontend/src/lib/components/chat/ChatInput.svelte @@ -95,6 +95,7 @@ onqueueedit?: (id: string) => void; onqueuedelete?: (id: string) => void; onsettingschange?: () => void; + ontoolapprovalchange?: (mode: ToolApprovalMode) => void; } let { inputText = $bindable(), @@ -123,7 +124,8 @@ onqueuesendnow, onqueueedit, onqueuedelete, - onsettingschange + onsettingschange, + ontoolapprovalchange }: Props = $props(); let editorEl: HTMLDivElement | undefined = $state(); @@ -1507,6 +1509,7 @@ bind:planMode bind:requestParams onchange={onsettingschange} + {ontoolapprovalchange} onfiles={(files) => { if (files) processFiles(Array.from(files)); }} diff --git a/cptr/frontend/src/lib/components/chat/ChatPanel.svelte b/cptr/frontend/src/lib/components/chat/ChatPanel.svelte index 33d7dd6a..9861ceca 100644 --- a/cptr/frontend/src/lib/components/chat/ChatPanel.svelte +++ b/cptr/frontend/src/lib/components/chat/ChatPanel.svelte @@ -35,7 +35,13 @@ import { socketStore } from '$lib/stores/socket.svelte'; import { onMount, onDestroy, tick } from 'svelte'; import { get } from 'svelte/store'; - import { currentWorkspace, openChatTab, streamingBehavior, widescreenMode } from '$lib/stores'; + import { + currentWorkspace, + openChatTab, + streamingBehavior, + toolApprovalMode as defaultToolApprovalMode, + widescreenMode + } from '$lib/stores'; import { getPathDisplayName } from '$lib/utils/paths'; import { ttsEnabled, @@ -663,7 +669,7 @@ const dm = get(defaultModel); if (dm) selectedModel = dm; else if (models.length) selectedModel = models[0].id; - toolApprovalMode = 'auto'; + toolApprovalMode = get(defaultToolApprovalMode); planMode = false; requestParams = {}; voiceModeEnabled = false; @@ -1062,6 +1068,12 @@ persistChatSettings(); } + function handleToolApprovalModeChange(mode: ToolApprovalMode) { + toolApprovalMode = mode; + defaultToolApprovalMode.set(mode); + persistChatSettings(); + } + function firstUserMessageTitle(): string { const message = allMessages.find((m) => m.role === 'user' && !isPendingHiddenMessage(m)); return message ? message.content.replace(/\s+/g, ' ').trim().slice(0, 80) : ''; @@ -1705,6 +1717,7 @@ onsend={send} onplan={handlePlanCommand} onsettingschange={persistChatSettings} + ontoolapprovalchange={handleToolApprovalModeChange} {queuedMessages} onqueuesendnow={handleQueueSendNow} onqueueedit={handleQueueEdit} @@ -1834,6 +1847,7 @@ onfork={handleForkChat} onplan={handlePlanCommand} onsettingschange={persistChatSettings} + ontoolapprovalchange={handleToolApprovalModeChange} onstatus={handleStatusCommand} onskillslist={handleSkillsListCommand} oncancel={handleCancel} diff --git a/cptr/frontend/src/lib/components/chat/PlusMenu.svelte b/cptr/frontend/src/lib/components/chat/PlusMenu.svelte index 70732053..6d22822a 100644 --- a/cptr/frontend/src/lib/components/chat/PlusMenu.svelte +++ b/cptr/frontend/src/lib/components/chat/PlusMenu.svelte @@ -13,6 +13,7 @@ planMode?: boolean; requestParams?: Record; onchange?: () => void; + ontoolapprovalchange?: (mode: ToolApprovalMode) => void; } let { onfiles, @@ -20,7 +21,8 @@ toolApprovalMode = $bindable('auto'), planMode = $bindable(false), requestParams = $bindable({}), - onchange + onchange, + ontoolapprovalchange }: Props = $props(); let open = $state(false); @@ -108,7 +110,8 @@ function selectMode(mode: ToolApprovalMode) { toolApprovalMode = mode; - onchange?.(); + if (ontoolapprovalchange) ontoolapprovalchange(mode); + else onchange?.(); } function triggerUpload() { diff --git a/cptr/frontend/src/lib/stores.ts b/cptr/frontend/src/lib/stores.ts index c5996693..1aa35105 100644 --- a/cptr/frontend/src/lib/stores.ts +++ b/cptr/frontend/src/lib/stores.ts @@ -110,11 +110,14 @@ export interface HomeState { splitDirection: SplitDirection; } +export type ToolApprovalMode = 'ask' | 'auto' | 'full'; + export interface UserPreferences { theme?: Theme; appearance?: AppearancePreferences; sidebarOpen: boolean; sidebarWidth: number; + toolApprovalMode?: ToolApprovalMode; locale: string; workspaceOrder?: string[]; // ordered paths for sidebar drag-reorder keybindings?: Record; // user-customised keyboard shortcuts @@ -298,6 +301,7 @@ if (typeof window !== 'undefined') { } export const sidebarWidth = writable(220); export const theme = writable('dark'); +export const toolApprovalMode = writable('auto'); export const appVersion = writable(''); export const lastSeenVersion = writable(''); export const latestVersion = writable(''); @@ -422,6 +426,7 @@ function persistPreferences(): void { }, sidebarOpen: get(sidebarOpen), sidebarWidth: get(sidebarWidth), + toolApprovalMode: get(toolApprovalMode), locale: i18next.language, workspaceOrder: get(workspaceOrder), keybindings: get(keybindings), @@ -459,6 +464,9 @@ function subscribeForPersistence() { sidebarWidth.subscribe(() => { if (get(stateLoaded)) persistPreferences(); }); + toolApprovalMode.subscribe(() => { + if (get(stateLoaded)) persistPreferences(); + }); workspaceOrder.subscribe(() => { if (get(stateLoaded)) persistPreferences(); }); @@ -500,6 +508,15 @@ export async function loadPreferences(): Promise { themeConfig.set(sanitizeThemeConfig(appearance.themeConfig)); if (prefs.sidebarOpen !== undefined) sidebarOpen.set(prefs.sidebarOpen as boolean); if (prefs.sidebarWidth !== undefined) sidebarWidth.set(prefs.sidebarWidth as number); + if ( + prefs.toolApprovalMode === 'ask' || + prefs.toolApprovalMode === 'auto' || + prefs.toolApprovalMode === 'full' + ) { + toolApprovalMode.set(prefs.toolApprovalMode); + } else if (prefs.autoApproveTools !== undefined) { + toolApprovalMode.set((prefs.autoApproveTools as boolean) ? 'full' : 'ask'); + } if (prefs.locale) changeLocale(prefs.locale as string); if (Array.isArray(prefs.workspaceOrder)) workspaceOrder.set(prefs.workspaceOrder as string[]); if (prefs.keybindings) loadKeybindings(prefs.keybindings as Record); From 72e9b35cb48fe531a930939357dc35fd09683a98 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 12 Jul 2026 01:03:03 -0500 Subject: [PATCH 04/12] refac --- cptr/routers/chat.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/cptr/routers/chat.py b/cptr/routers/chat.py index d3364374..aec71e8f 100644 --- a/cptr/routers/chat.py +++ b/cptr/routers/chat.py @@ -1055,7 +1055,14 @@ async def approve_tool(chat_id: str, message_id: str, body: ApproveRequest, requ result = await execute_tool( call["name"], call.get("arguments", {}), - {"workspace": chat.meta.get("workspace", ""), "user_id": user_id, "model_id": model_id}, + { + "workspace": chat.meta.get("workspace", ""), + "user_id": user_id, + "model_id": model_id, + "chat_id": chat_id, + "message_id": message_id, + "call_id": body.call_id, + }, ) call["status"] = "completed" output.append( From fc07e9bcad8dedf8ce9ba8c90b0d4288e4158392 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 12 Jul 2026 01:08:52 -0500 Subject: [PATCH 05/12] refac --- cptr/models/chats.py | 2 ++ cptr/utils/async_subagents.py | 55 ++++++++++++++++++++++++++--------- cptr/utils/tools.py | 1 - 3 files changed, 43 insertions(+), 15 deletions(-) diff --git a/cptr/models/chats.py b/cptr/models/chats.py index a954cd47..2b9b3f20 100644 --- a/cptr/models/chats.py +++ b/cptr/models/chats.py @@ -218,6 +218,8 @@ async def unread_counts_by_workspace( .where( Chat.user_id == user_id, workspace.in_(paths), + Chat.meta["internal"].as_boolean().is_not(True), + Chat.meta["subagent"].as_boolean().is_not(True), Chat.updated_at > func.coalesce(Chat.last_read_at, 0), ) .group_by(workspace) diff --git a/cptr/utils/async_subagents.py b/cptr/utils/async_subagents.py index c1988e26..cb2aa469 100644 --- a/cptr/utils/async_subagents.py +++ b/cptr/utils/async_subagents.py @@ -202,6 +202,7 @@ async def _inject_completion(record: dict[str, Any]) -> None: from cptr.utils.chat_task import get_pending_input_lock, start_task assistant_msg = None + direct_timer_completion = False async with get_pending_input_lock(parent_chat_id): all_msgs = await ChatMessage.get_all_by_chat(parent_chat_id) active = any(m.role == "assistant" and not m.done for m in all_msgs) @@ -216,29 +217,55 @@ async def _inject_completion(record: dict[str, Any]) -> None: if active: meta["async_subagent_pending"] = True - user_msg = await ChatMessage.create( - chat_id=parent_chat_id, - role="user", - content=content, - parent_id=parent_id, - model=model_id, - meta=meta, - created_at=now_ms(), - ) - - if not active: + if record.get("timer_chat_id") and not active: + direct_timer_completion = True + meta["timer_completion"] = True assistant_msg = await ChatMessage.create( chat_id=parent_chat_id, role="assistant", - content="", - parent_id=user_msg.id, + content=record.get("summary") or "Timer completed.", + parent_id=parent_id, model=model_id, - done=False, + done=True, + meta=meta, created_at=now_ms(), ) await Chat.update_current_message(parent_chat_id, assistant_msg.id, now_ms()) + else: + user_msg = await ChatMessage.create( + chat_id=parent_chat_id, + role="user", + content=content, + parent_id=parent_id, + model=model_id, + meta=meta, + created_at=now_ms(), + ) + + if not active: + assistant_msg = await ChatMessage.create( + chat_id=parent_chat_id, + role="assistant", + content="", + parent_id=user_msg.id, + model=model_id, + done=False, + created_at=now_ms(), + ) + await Chat.update_current_message(parent_chat_id, assistant_msg.id, now_ms()) await export_chat_to_file(parent_chat_id) + if direct_timer_completion: + await emit_to_user( + user_id, + { + "chat_id": parent_chat_id, + "message_id": assistant_msg.id, + "pending_inputs_processed": True, + }, + ) + return + if not assistant_msg: await emit_to_user( user_id, diff --git a/cptr/utils/tools.py b/cptr/utils/tools.py index bd8e11f8..cb65f09c 100644 --- a/cptr/utils/tools.py +++ b/cptr/utils/tools.py @@ -2620,7 +2620,6 @@ async def _run_subagent_chat( *BUILTIN_TOOL_GROUPS["browser"], *BUILTIN_TOOL_GROUPS["automations"], *BUILTIN_TOOL_GROUPS["images"], - *BUILTIN_TOOL_GROUPS["subagents"], "manage_skill", } From cd8761c2f4879590aef32ba4de5913839d1f45e5 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 12 Jul 2026 01:11:22 -0500 Subject: [PATCH 06/12] refac --- cptr/utils/timers.py | 16 +++++++++++----- cptr/utils/tools.py | 8 ++++++-- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/cptr/utils/timers.py b/cptr/utils/timers.py index 6fc3183b..fcd29ba3 100644 --- a/cptr/utils/timers.py +++ b/cptr/utils/timers.py @@ -12,7 +12,9 @@ logger = logging.getLogger(__name__) -_RELATIVE_TIME = re.compile(r"^\+(\d+)([smhd])$") +_RELATIVE_TIME = re.compile( + r"^(?:\+|in\s+)?(\d+)\s*(s|sec(?:onds?)?|m|min(?:utes?)?|h|hours?|d|days?)$" +) _RFC3339_TIME = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$") _TIME_UNITS_NS = { "s": 1_000_000_000, @@ -27,20 +29,24 @@ def parse_timer_at(value: str) -> int: """Normalize a relative offset or timezone-aware RFC 3339 timestamp.""" raw = value.strip() now = time.time_ns() - relative = _RELATIVE_TIME.fullmatch(raw) + relative = _RELATIVE_TIME.fullmatch(raw.lower()) if relative: count = int(relative.group(1)) if count <= 0: raise ValueError("at must be in the future.") - return now + count * _TIME_UNITS_NS[relative.group(2)] + return now + count * _TIME_UNITS_NS[relative.group(2)[0]] if not _RFC3339_TIME.fullmatch(raw): - raise ValueError("at must be +5s, +5m, +1h, +2d, or an RFC 3339 timestamp with a timezone.") + raise ValueError( + "at must be a relative time such as 10s or in 10 seconds, " + "or an RFC 3339 timestamp with a timezone." + ) try: parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) except ValueError as exc: raise ValueError( - "at must be +5s, +5m, +1h, +2d, or an RFC 3339 timestamp with a timezone." + "at must be a relative time such as 10s or in 10 seconds, " + "or an RFC 3339 timestamp with a timezone." ) from exc if parsed.tzinfo is None or parsed.utcoffset() is None: raise ValueError("absolute at values must include an explicit timezone.") diff --git a/cptr/utils/tools.py b/cptr/utils/tools.py index cb65f09c..104f347e 100644 --- a/cptr/utils/tools.py +++ b/cptr/utils/tools.py @@ -2379,10 +2379,14 @@ async def timer( """Set a one-shot timer for this chat. Use it when time is the missing input: wait for a deployment, retry after - backoff, revisit work after a deadline, or follow up after silence. At - `at`, the agent receives `prompt` and the latest thread context; it may + backoff, revisit work after a deadline, or follow up after silence. When + it fires, the agent receives `prompt` and the latest thread context; it may act, set another timer, or finish silently. + `at` is normally relative to now. Prefer `10s`, `5m`, `1h`, or `2d`; plain + language such as `in 10 seconds` also works. Use an RFC 3339 timestamp + with a timezone only for a real calendar deadline. + `cancel_on` prevents launch when `chat.read` or `chat.user_message` happens first in this chat. Omit it when timed work must run regardless. """ From 4a9499a60d48260ddac4eee3f5967bfab1231aed Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 12 Jul 2026 01:22:39 -0500 Subject: [PATCH 07/12] refac --- cptr/env.py | 7 +- .../components/chat/AssistantMessage.svelte | 14 +- .../src/lib/components/chat/ChatPanel.svelte | 9 +- cptr/utils/async_subagents.py | 8 - cptr/utils/timers.py | 196 ++++++------------ cptr/utils/tools.py | 10 - 6 files changed, 81 insertions(+), 163 deletions(-) diff --git a/cptr/env.py b/cptr/env.py index 4cc33351..5d856a58 100644 --- a/cptr/env.py +++ b/cptr/env.py @@ -20,6 +20,7 @@ def _env_int(name: str, default: int) -> int: except ValueError: return default + # ── Data directory ────────────────────────────────────────── # Where cptr stores its database, config, and user data. # Default: ~/.cptr @@ -32,9 +33,7 @@ def _env_int(name: str, default: int) -> int: LOG_FORMAT = os.environ.get("CPTR_LOG_FORMAT", "text").lower() AUDIT_LOG_LEVEL = os.environ.get("CPTR_AUDIT_LOG_LEVEL", "NONE").upper() -AUDIT_LOG_PATH = Path( - os.environ.get("CPTR_AUDIT_LOG_PATH", str(DATA_DIR / "logs" / "audit.jsonl")) -) +AUDIT_LOG_PATH = Path(os.environ.get("CPTR_AUDIT_LOG_PATH", str(DATA_DIR / "logs" / "audit.jsonl"))) AUDIT_LOG_ROTATION = os.environ.get("CPTR_AUDIT_LOG_ROTATION", "10 MB") AUDIT_MAX_BODY_SIZE = _env_int("CPTR_AUDIT_MAX_BODY_SIZE", 2048) AUDIT_EXCLUDED_PATHS = [ @@ -82,7 +81,7 @@ def _env_int(name: str, default: int) -> int: # ── Automation scheduler ──────────────────────────────────── AUTOMATION_POLL_INTERVAL = int(os.environ.get("AUTOMATION_POLL_INTERVAL", "10")) -TIMER_POLL_INTERVAL = int(os.environ.get("TIMER_POLL_INTERVAL", "10")) +TIMER_POLL_INTERVAL = int(os.environ.get("TIMER_POLL_INTERVAL", "1")) # ── CORS ──────────────────────────────────────────────────── # Socket.IO CORS allowed origins. diff --git a/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte b/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte index 0a8e9912..641e0318 100644 --- a/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte +++ b/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte @@ -16,8 +16,9 @@ import { t } from '$lib/i18n'; interface Props { - content: string; - done: boolean; + content: string; + meta?: Record | null; + done: boolean; output: any[] | null; usage: Record | null; chatId: string | null; @@ -34,8 +35,9 @@ onspeak?: () => void; } let { - content, - done, + content, + meta = null, + done, output, usage, chatId, @@ -59,6 +61,7 @@ let showUsageTooltip = $state(false); let collapsedFiles = $state>({}); let textareaEl: HTMLTextAreaElement; + const isTimer = $derived(meta?.timer === true); async function startEdit() { edit = true; @@ -450,6 +453,9 @@ {:else}
+ {#if isTimer} +
Timer
+ {/if} {#if !done && (!output || output.length === 0)} {#if !isLanding}
+
@@ -1755,7 +1759,7 @@
{#if hasHiddenMessages} @@ -1774,6 +1778,7 @@ {:else} None: await Chat.update_meta(timer.id, meta, now_ms()) -def _pack_parent_context(parent, messages, checkpoint_id: str | None) -> str: - """Build a small launch-time view of the parent without changing its history.""" - message_by_id = {message.id: message for message in messages} - branch = [] - current = message_by_id.get(parent.current_message_id) - while current: - branch.append(current) - current = message_by_id.get(current.parent_id) - branch.reverse() - - checkpoint_index = next( - (index for index, message in enumerate(branch) if message.id == checkpoint_id), -1 - ) - changes = branch[checkpoint_index + 1 :] if checkpoint_index >= 0 else branch[-12:] - changes = changes[-12:] - - parts = [] - summary = next( - (message.chat_summary for message in reversed(branch) if message.chat_summary), None - ) - if summary: - parts.append(f"## Parent summary\n{summary[-4_000:]}") - if changes: - lines = [] - remaining = 6_000 - for message in changes: - content = (message.content or "").strip() - if not content: - continue - content = content[: min(len(content), remaining)] - lines.append(f"{message.role}: {content}") - remaining -= len(content) - if remaining <= 0: - break - if lines: - parts.append("## Changes since timer was set\n" + "\n\n".join(lines)) - return "\n\n".join(parts) - - async def _set_timer_status(chat_id: str, status: str, **fields) -> None: from cptr.models import Chat from cptr.utils.config import now_ms @@ -134,31 +95,13 @@ async def _set_timer_status(chat_id: str, status: str, **fields) -> None: await Chat.update_meta(chat_id, meta, now_ms()) -async def set_timer_completion(record: dict, status: str, error: str | None) -> None: - """Persist completion before the normal async-subagent result reaches the parent.""" - timer_chat_id = record.get("timer_chat_id") - if not timer_chat_id: - return - final_status = "completed" if status == "completed" else "error" - await _set_timer_status( - timer_chat_id, - final_status, - timer_completed_at=time.time_ns(), - timer_error=error, - ) - - async def _launch_timer(timer, app) -> None: from cptr.models import Chat, ChatMessage - from cptr.utils.async_subagents import ( - attach_subagent_chat, - reserve_async_subagent, - start_async_subagent, - ) + from cptr.socket.main import emit_to_user from cptr.utils.chat_export import export_chat_to_file + from cptr.utils.chat_task import get_pending_input_lock, start_task from cptr.utils.config import now_ms - from cptr.utils.model_targets import ApiModelTarget, resolve_model_target - from cptr.utils.tools import _get_subagent_config, _run_existing_subagent_chat + from cptr.utils.model_targets import resolve_model_target async with _manager_lock: timer = await Chat.get_by_id(timer.id) @@ -173,93 +116,76 @@ async def _launch_timer(timer, app) -> None: await _set_timer_status(timer.id, "error", timer_error="parent chat no longer exists") return - config = await _get_subagent_config() - if not config["background_enabled"]: - return - - try: - target = await resolve_model_target(meta["timer_model_id"], app.state) - except Exception as exc: # model configuration can change while a timer waits - await _set_timer_status(timer.id, "error", timer_error=f"model unavailable: {exc}") - return - if not isinstance(target, ApiModelTarget): - await _set_timer_status( - timer.id, "error", timer_error="background timers require an API model" - ) - return - task_message = await ChatMessage.get_by_id(timer.current_message_id) if not task_message: await _set_timer_status(timer.id, "error", timer_error="timer task message is missing") return - reserve = await reserve_async_subagent( - config["max_async"], - task=task_message.content, - context="", - workspace=meta.get("workspace", ""), - user_id=timer.user_id, - parent_chat_id=parent.id, - parent_message_id=meta.get("timer_parent_message_id"), - connection=target.connection, - model=target.runtime_model, - model_id=target.full_model_id, - timer_chat_id=timer.id, - ) - if reserve.get("status") == "rejected": - return + async with get_pending_input_lock(parent.id): + parent_messages = await ChatMessage.get_all_by_chat(parent.id) + active = any( + message.role == "assistant" and not message.done for message in parent_messages + ) + if active: + return + + try: + target = await resolve_model_target(meta["timer_model_id"], app.state) + except Exception as exc: # model configuration can change while a timer waits + await _set_timer_status(timer.id, "error", timer_error=f"model unavailable: {exc}") + return + + done_assistants = [ + message + for message in parent_messages + if message.role == "assistant" and message.done + ] + parent_id = ( + done_assistants[-1].id if done_assistants else meta.get("timer_parent_message_id") + ) + assistant_msg = await ChatMessage.create( + chat_id=parent.id, + role="assistant", + content="", + parent_id=parent_id, + model=target.full_model_id, + done=False, + meta={"timer": True}, + created_at=now_ms(), + ) + await Chat.update_current_message(parent.id, assistant_msg.id, now_ms()) - parent_messages = await ChatMessage.get_all_by_chat(parent.id) - parent_context = _pack_parent_context( - parent, parent_messages, meta.get("timer_parent_message_id") - ) - content = task_message.content - if parent_context: - content = f"{content}\n\n## Parent context at launch\n{parent_context}" - await ChatMessage.update(task_message.id, content=content) - - assistant_msg = await ChatMessage.create( - chat_id=timer.id, - role="assistant", - content="", - parent_id=task_message.id, - model=target.full_model_id, - done=False, - created_at=now_ms(), - ) - meta.update( + meta.update( + { + "timer_status": "completed", + "timer_completed_at": time.time_ns(), + } + ) + await Chat.update_meta(timer.id, meta, now_ms()) + + await export_chat_to_file(timer.id) + await export_chat_to_file(parent.id) + + await emit_to_user( + timer.user_id, { - "timer_status": "running", - "timer_started_at": time.time_ns(), - "timer_assistant_message_id": assistant_msg.id, - "delegation_id": reserve["delegation_id"], - } + "chat_id": parent.id, + "message_id": assistant_msg.id, + "pending_inputs_processed": True, + }, ) - await Chat.update_meta(timer.id, meta, now_ms()) - await Chat.update_current_message(timer.id, assistant_msg.id, now_ms()) - await export_chat_to_file(timer.id) - await attach_subagent_chat( - reserve["delegation_id"], - subagent_chat_id=timer.id, - subagent_message_id=assistant_msg.id, + start_task( + message_id=assistant_msg.id, + chat_id=parent.id, + user_id=timer.user_id, + workspace=meta.get("workspace", ""), + target=target, + regeneration_prompt=task_message.content, ) - async def runner() -> str: - return await _run_existing_subagent_chat( - assistant_msg_id=assistant_msg.id, - chat_id=timer.id, - workspace=meta.get("workspace", ""), - connection=target.connection, - model=target.runtime_model, - user_id=timer.user_id, - config=config, - ) - - await start_async_subagent(reserve["delegation_id"], runner) - async def timer_worker_loop(app) -> None: - """Poll durable pending timers; a capacity miss simply waits for the next pass.""" + """Poll durable pending timers and wake their parent chats.""" from cptr.models import Chat logger.info("Timer worker started (poll interval: %ds)", TIMER_POLL_INTERVAL) diff --git a/cptr/utils/tools.py b/cptr/utils/tools.py index 104f347e..0e8ec0a7 100644 --- a/cptr/utils/tools.py +++ b/cptr/utils/tools.py @@ -2306,7 +2306,6 @@ async def delegate_task( model=__context__["model_id"], user_id=__context__["user_id"], parent_chat_id=__context__["chat_id"], - config=config, delegation_id=delegation_id, ) except Exception as e: @@ -2406,10 +2405,6 @@ async def timer( return "Error: cancel_on accepts only chat.read and chat.user_message." selected_events = list(dict.fromkeys(selected_events)) - config = await _get_subagent_config() - if not config["background_enabled"]: - return "Error: background sub-agents are disabled in settings." - full_model_id = __context__.get("full_model_id") or __context__["model_id"] chat, _, _ = await _create_subagent_chat( task=prompt, @@ -2418,7 +2413,6 @@ async def timer( model=full_model_id, user_id=__context__["user_id"], parent_chat_id=__context__["chat_id"], - config=config, child_type="timer", deferred=True, extra_meta={ @@ -2450,7 +2444,6 @@ async def _create_subagent_chat( model: str, user_id: str, parent_chat_id: str, - config: dict, delegation_id: str | None = None, *, child_type: str = "subagent", @@ -2560,7 +2553,6 @@ async def _run_subagent_chat( model=model, user_id=user_id, parent_chat_id=parent_chat_id, - config=config, ) return await _run_existing_subagent_chat( assistant_msg_id=assistant_msg.id, @@ -2979,8 +2971,6 @@ async def get_tool_list(builtin_tools: dict | None = None, workspace: str = "") "true", "1", ) - if not background_subagents_enabled: - tools.pop("timer", None) images_generation_enabled = (await Config.get("images.generation_enabled")) in ( True, "true", From fd7b3431b620f098c480a56bdcfe867e509a9e58 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 12 Jul 2026 01:22:57 -0500 Subject: [PATCH 08/12] refac --- cptr/frontend/src/lib/components/chat/ChatPanel.svelte | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cptr/frontend/src/lib/components/chat/ChatPanel.svelte b/cptr/frontend/src/lib/components/chat/ChatPanel.svelte index e5b1ed78..ba98e42a 100644 --- a/cptr/frontend/src/lib/components/chat/ChatPanel.svelte +++ b/cptr/frontend/src/lib/components/chat/ChatPanel.svelte @@ -1759,7 +1759,7 @@
{#if hasHiddenMessages} From f8d18fc71a85d6cf3e980800053d4a9cb2ef20e7 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 12 Jul 2026 01:26:21 -0500 Subject: [PATCH 09/12] refac --- .../components/chat/AssistantMessage.svelte | 19 +++++++++++-------- .../src/lib/components/chat/ChatPanel.svelte | 2 +- cptr/utils/timers.py | 2 +- 3 files changed, 13 insertions(+), 10 deletions(-) diff --git a/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte b/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte index 641e0318..efacddbd 100644 --- a/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte +++ b/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte @@ -16,9 +16,9 @@ import { t } from '$lib/i18n'; interface Props { - content: string; - meta?: Record | null; - done: boolean; + content: string; + meta?: Record | null; + done: boolean; output: any[] | null; usage: Record | null; chatId: string | null; @@ -35,9 +35,9 @@ onspeak?: () => void; } let { - content, - meta = null, - done, + content, + meta = null, + done, output, usage, chatId, @@ -61,7 +61,7 @@ let showUsageTooltip = $state(false); let collapsedFiles = $state>({}); let textareaEl: HTMLTextAreaElement; - const isTimer = $derived(meta?.timer === true); + const isTimer = $derived(meta?.internal === true && meta?.type === 'timer'); async function startEdit() { edit = true; @@ -454,7 +454,10 @@
{#if isTimer} -
Timer
+
+ Timer + +
{/if} {#if !done && (!output || output.length === 0)}
None: parent_id=parent_id, model=target.full_model_id, done=False, - meta={"timer": True}, + meta={"internal": True, "type": "timer"}, created_at=now_ms(), ) await Chat.update_current_message(parent.id, assistant_msg.id, now_ms()) From 48a4e6c3814199a905fc0abd8ae3427567cdf942 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 12 Jul 2026 01:30:49 -0500 Subject: [PATCH 10/12] refac --- .../components/chat/AssistantMessage.svelte | 7 ----- .../lib/components/chat/UserMessage.svelte | 30 ++++++++++++++++++- cptr/utils/timers.py | 13 ++++++-- 3 files changed, 39 insertions(+), 11 deletions(-) diff --git a/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte b/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte index efacddbd..f4e8f86c 100644 --- a/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte +++ b/cptr/frontend/src/lib/components/chat/AssistantMessage.svelte @@ -61,7 +61,6 @@ let showUsageTooltip = $state(false); let collapsedFiles = $state>({}); let textareaEl: HTMLTextAreaElement; - const isTimer = $derived(meta?.internal === true && meta?.type === 'timer'); async function startEdit() { edit = true; @@ -453,12 +452,6 @@ {:else}
- {#if isTimer} -
- Timer - -
- {/if} {#if !done && (!output || output.length === 0)}
- {#if isAsyncSubagentResult} + {#if isTimer} +
+ + {#if timerExpanded} +
+ {content} +
+ {/if} +
+ {:else if isAsyncSubagentResult}