diff --git a/coworker/agent.py b/coworker/agent.py index 9479718c8..6c8cabd38 100644 --- a/coworker/agent.py +++ b/coworker/agent.py @@ -247,6 +247,7 @@ def build_engine( # Persona-carried skill folders (OPE-58): the bundle's skills/ dir joins the loader so # its skills are readable by load_skill, not just listed by the filter. extra_skill_dirs: Optional[list[str | Path]] = None, + default_approval_ttl_seconds: Optional[float] = None, ) -> TurnEngine: ws = Path(workspace).expanduser().resolve() if workspace else None if agent.requires_folder and ws is None: @@ -548,6 +549,11 @@ def context_provider() -> str: tool_requester=tool_requester, team_approver=team_approver, items_approver=items_approver, + default_approval_ttl_seconds=( + default_approval_ttl_seconds + if default_approval_ttl_seconds is not None + else config.inbox_approval_ttl_seconds + ), ) engine.executor = executor # type: ignore[attr-defined] engine.todo = todo # type: ignore[attr-defined] diff --git a/coworker/config.py b/coworker/config.py index 43fe33fa8..d3004f7ea 100644 --- a/coworker/config.py +++ b/coworker/config.py @@ -73,6 +73,8 @@ class Config: cloud_relay_ws_url: str = ( "wss://l4z1paxb83.execute-api.us-east-1.amazonaws.com/ocw-connect" ) + # Default TTL (seconds) for unattended/parked approval items. None means no expiry. + inbox_approval_ttl_seconds: Optional[float] = None _FIELDS = { @@ -92,6 +94,7 @@ class Config: "cloud_client_id", "cloud_audience", "cloud_relay_ws_url", + "inbox_approval_ttl_seconds", } # These fields change what consequential actions can run without a prompt, so the normal diff --git a/coworker/engine.py b/coworker/engine.py index e492443c4..522ae129b 100644 --- a/coworker/engine.py +++ b/coworker/engine.py @@ -56,6 +56,7 @@ class ApprovalOutcome(str, Enum): # nothing persisted. EXTERNAL-risk tools only (validated server-side). THIS_RUN = "this_run" DENY = "deny" + EXPIRED = "expired" def _readonly_ok(arguments: dict) -> bool: @@ -78,6 +79,8 @@ class PermissionRequest: # registration) — carried on the request so a PARKED approval shows the same # destination evidence as the live card (§35 parity). None for non-MCP tools. mcp_destination: Optional[dict] = None + expires_at: Optional[str] = None + ttl_seconds: Optional[float] = None Approver = Callable[[PermissionRequest], Awaitable[ApprovalOutcome]] @@ -123,12 +126,14 @@ def __init__( # Called (thread-safe, best-effort) when the user stops the turn — e.g. the # executor's kill for a running shell command. interrupt_hooks: Optional[list[Callable[[], None]]] = None, + default_approval_ttl_seconds: Optional[float] = None, ) -> None: self.provider = provider self.registry = registry self.permissions = permissions self.model = model self.approver = approver or _deny_all + self.default_approval_ttl_seconds = default_approval_ttl_seconds self.max_iterations = max_iterations self.model_settings = dict(model_settings or {}) self.messages: list[dict[str, Any]] = list(messages or []) @@ -1131,6 +1136,7 @@ async def _authorize(self, tool_call: ToolCall) -> "AsyncIterator[Event | bool]" ) allowed = decision.allowed reason = decision.reason + outcome: Optional[ApprovalOutcome] = None # OPE-114 §1: running something the agent DOWNLOADED this session is the classic # fetch-then-execute chain, and there is no quiet legitimate version of it — so it @@ -1352,11 +1358,30 @@ async def _authorize(self, tool_call: ToolCall) -> "AsyncIterator[Event | bool]" if spec else None ), + ttl_seconds=self.default_approval_ttl_seconds, ) ), interrupted=ApprovalOutcome.DENY, ) - if outcome is ApprovalOutcome.DENY: + if outcome is ApprovalOutcome.EXPIRED: + allowed, reason = ( + False, + "approval request expired (TTL elapsed)", + ) + self._approval_origins[tool_call.id] = { + "origin": "timeout", + "grant": "expired", + **({"note": unsure_note} if unsure_note else {}), + } + self._audit( + tool_call, + stage="approval_resolved", + call_id=tool_call.id, + status="expired", + approval=outcome.value, + reason=reason, + ) + elif outcome is ApprovalOutcome.DENY: allowed, reason = ( False, "interrupted by user" if self._cancel.is_set() else "denied by user", @@ -1421,11 +1446,14 @@ async def _authorize(self, tool_call: ToolCall) -> "AsyncIterator[Event | bool]" **({"approval_grant": origin["grant"]} if origin.get("grant") else {}), } self.messages.append(err_msg) + finish_status = ( + "expired" if outcome is ApprovalOutcome.EXPIRED else "denied" + ) yield Event( EventType.TOOL_FINISHED, - {"name": tool_call.name, "status": "denied", "reason": reason}, + {"name": tool_call.name, "status": finish_status, "reason": reason}, ) - self._audit(tool_call, stage="finished", status="denied", reason=reason) + self._audit(tool_call, stage="finished", status=finish_status, reason=reason) yield False return diff --git a/coworker/inbox.py b/coworker/inbox.py index 4491227e8..e13026c5b 100644 --- a/coworker/inbox.py +++ b/coworker/inbox.py @@ -18,7 +18,7 @@ import threading import uuid from dataclasses import asdict, dataclass, field -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Optional @@ -44,6 +44,21 @@ def _now() -> str: return datetime.now(timezone.utc).isoformat() +def is_expired(item: InboxItem, now: Optional[datetime] = None) -> bool: + """True if the item has an expires_at timestamp that is in the past.""" + if not item.expires_at: + return False + if now is None: + now = datetime.now(timezone.utc) + try: + exp = datetime.fromisoformat(item.expires_at) + if exp.tzinfo is None: + exp = exp.replace(tzinfo=timezone.utc) + return now >= exp + except (ValueError, TypeError): + return False + + def args_preview(arguments: Optional[dict], *, limit: int = 240) -> str: """A compact one-line summary of a tool call's arguments, for an approval card body (so a mirrored 'Run `write_file`?' shows *what* — path/content — not just the tool name). @@ -68,7 +83,7 @@ class InboxItem: body: str = "" state: str = STATE_PENDING resolution: Optional[str] = ( - None # approval: "allow"/"deny"/"always"; question: answer text + None # approval: "allow"/"deny"/"always"/"expired"; question: answer text ) inbox: str = "default" # named inbox / delivery binding (Phase 3 routing) created_at: str = field(default_factory=_now) @@ -94,11 +109,19 @@ class InboxItem: questions: list[dict] = field(default_factory=list) # Kind-specific payload (directory: suggested path/writable; plan: the plan text; …). data: dict[str, Any] = field(default_factory=dict) + # Optional expiry timestamp (ISO-8601 UTC). When elapsed, item auto-resolves as "expired". + expires_at: Optional[str] = None class InboxStore: - def __init__(self, path: Optional[str | Path] = None) -> None: + def __init__( + self, + path: Optional[str | Path] = None, + *, + default_ttl_seconds: Optional[float] = None, + ) -> None: self.path = Path(path) if path else None + self.default_ttl_seconds = default_ttl_seconds self._lock = threading.Lock() self._items: dict[str, InboxItem] = {} self._waiters: dict[str, asyncio.Event] = {} @@ -121,6 +144,18 @@ def _save(self) -> None: encoding="utf-8", ) + def _compute_expires_at( + self, + expires_at: Optional[str] = None, + ttl_seconds: Optional[float] = None, + ) -> Optional[str]: + if expires_at: + return expires_at + ttl = ttl_seconds if ttl_seconds is not None else self.default_ttl_seconds + if ttl is not None and ttl > 0: + return (datetime.now(timezone.utc) + timedelta(seconds=ttl)).isoformat() + return None + # -- adding ----------------------------------------------------------------- def add( self, @@ -138,6 +173,8 @@ def add( header: str = "", questions=None, tool_call_id: Optional[str] = None, + expires_at: Optional[str] = None, + ttl_seconds: Optional[float] = None, ) -> InboxItem: # Idempotent by (session_id, tool_call_id): a durable resume re-raises the same prompt, and # must reuse the existing (possibly already-resolved) item rather than re-prompt. @@ -145,6 +182,7 @@ def add( existing = self.for_tool_call(session_id, tool_call_id) if existing is not None: return existing + computed_expires_at = self._compute_expires_at(expires_at, ttl_seconds) item = InboxItem( id=uuid.uuid4().hex, session_id=session_id, @@ -160,6 +198,7 @@ def add( header=str(header or ""), questions=list(questions or []), tool_call_id=tool_call_id, + expires_at=computed_expires_at, ) with self._lock: self._items[item.id] = item @@ -182,6 +221,8 @@ def add_approval( visibility=VIS_INBOX, data=None, tool_call_id=None, + expires_at=None, + ttl_seconds=None, ) -> InboxItem: # `data` carries the automation-run context for standing scoped approvals (§25): # {task_id, task_title, standing_target?} — the in-app card's "Allow every time" gate. @@ -194,6 +235,8 @@ def add_approval( visibility=visibility, data=data, tool_call_id=tool_call_id, + expires_at=expires_at, + ttl_seconds=ttl_seconds, ) def add_question( @@ -210,6 +253,8 @@ def add_question( header="", questions=None, tool_call_id=None, + expires_at=None, + ttl_seconds=None, ) -> InboxItem: return self.add( session_id, @@ -224,6 +269,8 @@ def add_question( header=header, questions=questions, tool_call_id=tool_call_id, + expires_at=expires_at, + ttl_seconds=ttl_seconds, ) def add_directory( @@ -236,6 +283,8 @@ def add_directory( visibility=VIS_INBOX, data=None, tool_call_id=None, + expires_at=None, + ttl_seconds=None, ) -> InboxItem: return self.add( session_id, @@ -246,6 +295,8 @@ def add_directory( visibility=visibility, data=data, tool_call_id=tool_call_id, + expires_at=expires_at, + ttl_seconds=ttl_seconds, ) def add_plan( @@ -258,6 +309,8 @@ def add_plan( visibility=VIS_INBOX, data=None, tool_call_id=None, + expires_at=None, + ttl_seconds=None, ) -> InboxItem: return self.add( session_id, @@ -268,6 +321,8 @@ def add_plan( visibility=visibility, data=data, tool_call_id=tool_call_id, + expires_at=expires_at, + ttl_seconds=ttl_seconds, ) def add_tool_request( @@ -280,6 +335,8 @@ def add_tool_request( visibility=VIS_INBOX, data=None, tool_call_id=None, + expires_at=None, + ttl_seconds=None, ) -> InboxItem: return self.add( session_id, @@ -290,10 +347,20 @@ def add_tool_request( visibility=visibility, data=data, tool_call_id=tool_call_id, + expires_at=expires_at, + ttl_seconds=ttl_seconds, ) def add_notification( - self, session_id, title, *, body="", inbox="default", visibility=VIS_INBOX + self, + session_id, + title, + *, + body="", + inbox="default", + visibility=VIS_INBOX, + expires_at=None, + ttl_seconds=None, ) -> InboxItem: return self.add( session_id, @@ -302,11 +369,37 @@ def add_notification( body=body, inbox=inbox, visibility=visibility, + expires_at=expires_at, + ttl_seconds=ttl_seconds, ) # -- queries ---------------------------------------------------------------- + def _check_expirations_locked(self) -> list[InboxItem]: + """Check all pending items and auto-resolve any whose TTL has elapsed.""" + expired: list[InboxItem] = [] + now = datetime.now(timezone.utc) + for item in self._items.values(): + if item.state == STATE_PENDING and is_expired(item, now=now): + item.state = STATE_RESOLVED + item.resolution = "expired" + item.resolved_at = now.isoformat() + expired.append(item) + ev = self._waiters.get(item.id) + if ev is not None: + ev.set() + if expired: + self._save() + return expired + + def check_expirations(self) -> list[InboxItem]: + """Check all pending items and auto-resolve any whose TTL has elapsed.""" + with self._lock: + return self._check_expirations_locked() + def get(self, item_id: str) -> Optional[InboxItem]: - return self._items.get(item_id) + with self._lock: + self._check_expirations_locked() + return self._items.get(item_id) def list( self, @@ -316,7 +409,9 @@ def list( inbox: Optional[str] = None, visibility: Optional[str] = None, ) -> list[InboxItem]: - out = list(self._items.values()) + with self._lock: + self._check_expirations_locked() + out = list(self._items.values()) if session_id is not None: out = [i for i in out if i.session_id == session_id] if state is not None: @@ -331,13 +426,23 @@ def pending(self, session_id: Optional[str] = None) -> list[InboxItem]: return self.list(session_id=session_id, state=STATE_PENDING) # -- the state machine ------------------------------------------------------ - def resolve(self, item_id: str, resolution: str) -> bool: + def resolve(self, item_id: str, resolution: str, *, force: bool = False) -> bool: """Resolve an item exactly once. First responder wins; later attempts are no-ops - (return False). Fires any awaiting agent (the suspended inbox_approver).""" + (return False). If the item's TTL has elapsed, resolving with a user decision is + rejected: the item auto-resolves as 'expired' to prevent acting on stale consent.""" with self._lock: item = self._items.get(item_id) if item is None or item.state == STATE_RESOLVED: return False + if not force and is_expired(item): + item.state = STATE_RESOLVED + item.resolution = "expired" + item.resolved_at = _now() + self._save() + waiter = self._waiters.get(item_id) + if waiter is not None: + waiter.set() + return False item.state = STATE_RESOLVED item.resolution = resolution item.resolved_at = _now() @@ -355,20 +460,42 @@ def resolve_session( the usual way; returns how many items were closed.""" closed = 0 for item in self.pending(session_id): - if self.resolve(item.id, resolution): + if self.resolve(item.id, resolution, force=True): closed += 1 return closed async def wait(self, item_id: str) -> str: """Await an item's resolution; returns the resolution string. Used by the approver to - suspend the agent until a human answers (from any surface).""" - item = self._items.get(item_id) - if item is not None and item.state == STATE_RESOLVED: - return item.resolution or "" - ev = self._waiters.setdefault(item_id, asyncio.Event()) - await ev.wait() - resolved = self._items.get(item_id) - return (resolved.resolution if resolved else "") or "" + suspend the agent until a human answers (from any surface). If the item has an expiry + and elapses before resolution, it auto-resolves as 'expired'.""" + with self._lock: + self._check_expirations_locked() + item = self._items.get(item_id) + if item is not None and item.state == STATE_RESOLVED: + return item.resolution or "" + timeout: Optional[float] = None + if item is not None and item.expires_at: + try: + exp = datetime.fromisoformat(item.expires_at) + if exp.tzinfo is None: + exp = exp.replace(tzinfo=timezone.utc) + timeout = max(0.0, (exp - datetime.now(timezone.utc)).total_seconds()) + except (ValueError, TypeError): + pass + ev = self._waiters.setdefault(item_id, asyncio.Event()) + + try: + if timeout is not None: + await asyncio.wait_for(ev.wait(), timeout=timeout) + else: + await ev.wait() + except asyncio.TimeoutError: + # Resolve outside the lock and preserve a concurrent first response. + self.resolve(item_id, "expired", force=True) + + with self._lock: + resolved = self._items.get(item_id) + return (resolved.resolution if resolved else "") or "" # -- resume reconciliation -------------------------------------------------- def reconcile_on_resume(self, session_id: str) -> dict: @@ -386,7 +513,7 @@ def reconcile_on_resume(self, session_id: str) -> dict: # -- approver routing ----------------------------------------------------------- def inbox_approver(store: InboxStore, session_id: str, *, inbox: str = "default"): """An Approver that routes a permission request to the Inbox and suspends until resolved. - Maps the resolution to an ApprovalOutcome (allow → ONCE, always → ALWAYS_TOOL, else DENY). + Maps the resolution to an ApprovalOutcome (allow → ONCE, always → ALWAYS_TOOL, expired → EXPIRED, else DENY). """ from .engine import ApprovalOutcome, PermissionRequest @@ -396,12 +523,16 @@ async def approve(request: "PermissionRequest") -> "ApprovalOutcome": title=f"Run `{request.tool_name}`?", body=request.reason or "", inbox=inbox, + expires_at=getattr(request, "expires_at", None), + ttl_seconds=getattr(request, "ttl_seconds", None), ) resolution = await store.wait(item.id) if resolution == "always": return ApprovalOutcome.ALWAYS_TOOL if resolution == "allow": return ApprovalOutcome.ONCE + if resolution == "expired": + return ApprovalOutcome.EXPIRED return ApprovalOutcome.DENY return approve diff --git a/coworker/server/manager.py b/coworker/server/manager.py index cf9b740fa..926783c83 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -318,7 +318,10 @@ def __init__( set_persona_registry(self.personas) # Inbox (cross-session human-attention queue), routing (named inboxes + Slack/Telegram # bindings), the Unattended toggle, and self-wake records. - self.inbox = InboxStore(base / "inbox.json") + self.inbox = InboxStore( + base / "inbox.json", + default_ttl_seconds=load_config().inbox_approval_ttl_seconds, + ) self.inbox_routing = InboxRouting(base / "inbox_routing.json") self.unattended = UnattendedRegistry(base / "unattended.json") self.wakes = WakeStore(base / "wakes.json") @@ -1053,6 +1056,8 @@ async def ask( session_id, inbox=inbox_name, tool_call_id=tool_call_id, + expires_at=args.get("expires_at"), + ttl_seconds=args.get("ttl_seconds"), **fields, ) if ( @@ -1079,6 +1084,8 @@ async def approve(request): inbox=self.inbox_routing.route_for(session_id, agent), tool_call_id=getattr(request, "tool_call_id", None), data=self.approval_prompt_data(session_id, request), + expires_at=getattr(request, "expires_at", None), + ttl_seconds=getattr(request, "ttl_seconds", None), ) if item.state == "pending": self.persist_session(session_id) @@ -1101,11 +1108,16 @@ async def request(args, tool_call_id=None): "primary": bool(args.get("primary", False)), }, tool_call_id=tool_call_id, + expires_at=args.get("expires_at"), + ttl_seconds=args.get("ttl_seconds"), ) if item.state == "pending": self.persist_session(session_id) await self.mirror_inbox_item(item) - resp = _parse_inbox_json(await self.inbox.wait(item.id)) + resolution = await self.inbox.wait(item.id) + if resolution == "expired": + return {"granted": False, "reason": "the request expired (TTL elapsed)"} + resp = _parse_inbox_json(resolution) if not resp.get("granted"): return {"granted": False, "reason": "the user declined the request"} path = (resp.get("path") or args.get("path") or "").strip() @@ -1157,11 +1169,19 @@ async def approve(args, tool_call_id=None): body=str(args.get("plan", "")), inbox=self.inbox_routing.route_for(session_id, agent), tool_call_id=tool_call_id, + expires_at=args.get("expires_at"), + ttl_seconds=args.get("ttl_seconds"), ) if item.state == "pending": self.persist_session(session_id) await self.mirror_inbox_item(item) - resp = _parse_inbox_json(await self.inbox.wait(item.id)) + resolution = await self.inbox.wait(item.id) + if resolution == "expired": + return { + "approved": False, + "feedback": "the plan request expired (TTL elapsed)", + } + resp = _parse_inbox_json(resolution) if not resp.get("approved"): return { "approved": False, @@ -4321,6 +4341,8 @@ def approval_outcome(self, resolution: str, request, session_id: str): if not minted: self._audit_grant_refused(session_id, request, resolution) return ApprovalOutcome.ONCE + if resolution == "expired": + return ApprovalOutcome.EXPIRED try: outcome = ApprovalOutcome(resolution) except ValueError: @@ -4434,6 +4456,8 @@ async def approver(request): inbox=self.inbox_routing.route_for(session_id, task.agent), tool_call_id=getattr(request, "tool_call_id", None), data=self.approval_prompt_data(session_id, request), + expires_at=getattr(request, "expires_at", None), + ttl_seconds=getattr(request, "ttl_seconds", None), ) if item.state == "pending": self.persist_session(session_id) @@ -4598,15 +4622,32 @@ def _resolve(item_id: str, resolution: str) -> bool: # -- self-wake resumption --------------------------------------------------- async def _scheduler_tick(self) -> None: - """The shared per-tick work: resume due self-wakes, then drain team queues. - Team deliveries dispatch as tasks (a long worker turn must not stall the - scheduler).""" + """The shared per-tick work: resume due self-wakes, expire due inbox items, + then drain team queues. Team deliveries dispatch as tasks (a long worker + turn must not stall the scheduler).""" await self.resume_due_wakes() + try: + await self.expire_due_inbox_items() + except Exception: + logger.exception("inbox expiration check failed") try: await self.team_tick() except Exception: logger.exception("team tick failed") + async def expire_due_inbox_items(self) -> int: + """Auto-resolve pending inbox items whose TTL has expired and resume parked sessions.""" + expired = self.inbox.check_expirations() + for item in expired: + if not self.is_running(item.session_id): + try: + await self._durable_resume(item) + except Exception: + logger.exception( + "failed to resume session %s after item expiry", item.session_id + ) + return len(expired) + async def resume_due_wakes(self) -> int: """Resume sessions whose self-wakes are due (called each scheduler tick). A suspended agent (it called sleep_until / wake_on / wake_on_event and ended its turn) is re-invoked on diff --git a/surfaces/gui/src/api.ts b/surfaces/gui/src/api.ts index a69e0ceca..3990a72d7 100644 --- a/surfaces/gui/src/api.ts +++ b/surfaces/gui/src/api.ts @@ -1549,6 +1549,7 @@ export interface InboxItem { session_agent?: string | null; session_workspace?: string | null; session_exists?: boolean; + expires_at?: string | null; } export async function getInbox(sessionId?: string, state?: string): Promise { diff --git a/tests/test_engine.py b/tests/test_engine.py index 3c4c24e2f..27db43a1d 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -475,3 +475,49 @@ def test_ordinary_text_answer_still_completes(tmp_path): events = _collect(engine, "how does qwen format tool calls?") assert EventType.ERROR not in _types(events) assert next(ev for ev in events if ev.type == EventType.TURN_END).data["status"] == "completed" + + +def test_turn_engine_approval_expired_resumes_with_notice(tmp_path): + audits = [] + + async def _approver(req: PermissionRequest) -> ApprovalOutcome: + return ApprovalOutcome.EXPIRED + + engine, _ = _engine( + tmp_path, + [ + _tool_turn("write_file", {"path": "test.txt", "content": "hello"}), + _text_turn("Approval lapsed, so I did not write the file."), + ], + approver=_approver, + ) + engine.audit_sink = audits.append + + events = _collect(engine, "write the file") + + tool_fins = [ev for ev in events if ev.type == EventType.TOOL_FINISHED] + assert len(tool_fins) == 1 + assert tool_fins[0].data["status"] == "expired" + assert "approval request expired" in tool_fins[0].data["reason"] + assert any( + m.get("role") == "tool" and "approval request expired" in m["content"] + for m in engine.messages + ) + assert any( + a.get("stage") == "approval_resolved" and a.get("status") == "expired" + for a in audits + ) + assert any(ev.type == EventType.TURN_END for ev in events) + + +async def test_plan_mode_hard_denial_does_not_require_approval_outcome(tmp_path): + from coworker.permissions import Mode + + engine, _ = _engine(tmp_path, []) + engine.permissions.mode = Mode.PLAN + events = [e async for e in engine._authorize( + ToolCall(id="denied", name="write_file", arguments={"path": "file.txt", "content": "x"}) + )] + assert events[-1] is False + assert any(getattr(e, "data", {}).get("status") == "denied" for e in events) + assert not (tmp_path / "file.txt").exists() diff --git a/tests/test_inbox.py b/tests/test_inbox.py index e4c97120a..085001024 100644 --- a/tests/test_inbox.py +++ b/tests/test_inbox.py @@ -179,3 +179,100 @@ def test_approval_prompt_data_carries_mcp_evidence(): plain = PermissionRequest("write_file", {"path": "g.txt"}, None, "requires approval") plain_data = SessionManager.approval_prompt_data(fake_mgr, "s1", plain) assert "mcp_destination" not in plain_data and "category" not in plain_data + + +def test_inbox_item_expiry_calculation_and_is_expired(tmp_path): + from datetime import datetime, timedelta, timezone + + from coworker.inbox import is_expired + + store = InboxStore(tmp_path / "inbox.json") + item_ttl = store.add_approval("s1", "Deploy?", ttl_seconds=60) + assert item_ttl.expires_at is not None + assert is_expired(item_ttl) is False + + past = (datetime.now(timezone.utc) - timedelta(seconds=10)).isoformat() + item_past = store.add_approval("s1", "Old request", expires_at=past) + assert is_expired(item_past) is True + + +def test_inbox_store_default_ttl(tmp_path): + store = InboxStore(tmp_path / "inbox.json", default_ttl_seconds=120) + item = store.add_approval("s1", "Deploy?") + assert item.expires_at is not None + + # Explicit ttl overrides default + item2 = store.add_approval("s1", "Fast", ttl_seconds=10) + assert item2.expires_at is not None + + +def test_inbox_resolve_expired_item_rejected(tmp_path): + from datetime import datetime, timedelta, timezone + + store = InboxStore(tmp_path / "inbox.json") + past = (datetime.now(timezone.utc) - timedelta(seconds=5)).isoformat() + item = store.add_approval("s1", "Old request", expires_at=past) + + # Late resolve attempt fails and marks resolution as expired + ok = store.resolve(item.id, "allow") + assert ok is False + assert item.state == STATE_RESOLVED + assert item.resolution == "expired" + + +def test_inbox_list_and_get_auto_expire(tmp_path): + from datetime import datetime, timedelta, timezone + + store = InboxStore(tmp_path / "inbox.json") + past = (datetime.now(timezone.utc) - timedelta(seconds=5)).isoformat() + item = store.add_approval("s1", "Old request", expires_at=past) + + # Pending list does not include expired item + assert len(store.pending("s1")) == 0 + + # get() returns it resolved as expired + got = store.get(item.id) + assert got.state == STATE_RESOLVED + assert got.resolution == "expired" + + +def test_inbox_wait_auto_resolves_when_ttl_elapses(tmp_path): + async def run(): + store = InboxStore(tmp_path / "inbox.json") + item = store.add_approval("s1", "Quick TTL", ttl_seconds=0.05) + res = await store.wait(item.id) + assert res == "expired" + assert item.state == STATE_RESOLVED + assert item.resolution == "expired" + + asyncio.run(run()) + + +def test_inbox_approver_expired_outcome(tmp_path): + async def run(): + store = InboxStore(tmp_path / "inbox.json") + from coworker.engine import ApprovalOutcome, PermissionRequest + + approver = inbox_approver(store, "s1") + req = PermissionRequest("run_shell", {}, None, "needs approval", ttl_seconds=0.05) + outcome = await approver(req) + assert outcome is ApprovalOutcome.EXPIRED + + asyncio.run(run()) + + + +def test_wait_already_expired_does_not_deadlock(): + import subprocess + import sys + from pathlib import Path + + result = subprocess.run([sys.executable, "-c", """ +import asyncio +from coworker.inbox import InboxStore +s = InboxStore() +i = s.add_approval('session', 'approval', expires_at='2000-01-01T00:00:00+00:00') +assert asyncio.run(s.wait(i.id)) == 'expired' +assert s.get(i.id).resolution == 'expired' +"""], cwd=Path(__file__).resolve().parents[1], capture_output=True, text=True, timeout=5) + assert result.returncode == 0, result.stderr