#!/usr/bin/env python3 """Chat-scoped scheduled-job CLI for AgentsDock agent turns.""" from __future__ import annotations import argparse import ipaddress import json import os import re import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any PROVIDER_RUNTIME_VALUE_MAX_BYTES = 4096 class JobsCLIError(RuntimeError): """A safe, user-facing CLI failure.""" class NoRedirectHandler(urllib.request.HTTPRedirectHandler): def redirect_request(self, *_args: Any, **_kwargs: Any) -> None: return None def host_is_loopback(host: str) -> bool: try: address = ipaddress.ip_address(host) except ValueError: return host.lower() == "localhost" if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped: address = address.ipv4_mapped return address.is_loopback def canonical_http_origin(value: str, label: str) -> tuple[str, bool]: raw = value.strip() try: parsed = urllib.parse.urlsplit(raw) port = parsed.port or 80 except ValueError as exc: raise JobsCLIError(f"{label} must be an HTTP origin") from exc if ( parsed.scheme.lower() != "http" or not parsed.hostname or parsed.username is not None or parsed.password is not None or parsed.path not in {"", "/"} or parsed.query or parsed.fragment ): raise JobsCLIError(f"{label} must be an HTTP origin") host = parsed.hostname.lower() try: address = ipaddress.ip_address(host) if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped: address = address.ipv4_mapped host = address.compressed loopback = address.is_loopback url_host = f"[{host}]" if isinstance(address, ipaddress.IPv6Address) else host except ValueError: loopback = host == "localhost" url_host = host return f"http://{url_host}:{port}", loopback def nonempty_chat_id(value: str) -> str: chat_id = value.strip() if not chat_id: raise argparse.ArgumentTypeError("--chat-id must not be empty") return chat_id def chat_route_selection(value: str) -> dict[str, str]: """Parse an opaque live route without accepting an internal chat id.""" raw = value.strip() route_id, separator, action = raw.partition("=") if not re.fullmatch(r"route_[0-9a-f]{32}", route_id): raise argparse.ArgumentTypeError( "--chat-route must use an opaque route ID returned by the chats helper" ) selected_action = action.strip() if separator else "instruction" if selected_action not in {"instruction", "request_reply"}: raise argparse.ArgumentTypeError( "--chat-route action must be instruction or request_reply" ) return {"route_id": route_id, "action": selected_action} def bounded_identity_value(value: str | None, label: str) -> str: clean = str(value or "").strip() try: size = len(clean.encode("utf-8")) except UnicodeEncodeError as exc: raise JobsCLIError(f"{label} is not valid UTF-8") from exc if size > PROVIDER_RUNTIME_VALUE_MAX_BYTES: raise JobsCLIError(f"{label} exceeds the provider runtime limit") return clean def selected_authority_path(authority_file: str | None = None) -> Path: explicit = bounded_identity_value(authority_file, "--authority-file") ambient = bounded_identity_value( os.environ.get("AGENTSDOCK_PROVIDER_AUTHORITY_FILE"), "AGENTSDOCK_PROVIDER_AUTHORITY_FILE", ) if explicit and ambient: explicit_key = os.path.abspath(os.path.expanduser(explicit)) ambient_key = os.path.abspath(os.path.expanduser(ambient)) if explicit_key != ambient_key: raise JobsCLIError( "--authority-file conflicts with the live provider authority" ) selected = explicit or ambient if not selected: raise JobsCLIError("--authority-file is required") return Path(selected).expanduser() def provider_authority(authority_file: str | None = None) -> tuple[str, str]: path = selected_authority_path(authority_file) try: if path.stat().st_mode & 0o077: raise JobsCLIError("authority file permissions are unsafe") payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise JobsCLIError(f"could not read authority file: {exc}") from exc capability = str(payload.get("provider_capability") or payload.get("capability") or "") source_session_id = str(payload.get("source_session_id") or "").strip() if not capability or not source_session_id: raise JobsCLIError("authority file is invalid") return capability, source_session_id def authority_server_origin(authority_file: str | None = None) -> str: path = selected_authority_path(authority_file) try: if path.stat().st_mode & 0o077: raise JobsCLIError("authority file permissions are unsafe") payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise JobsCLIError(f"could not read authority file: {exc}") from exc return bounded_identity_value( payload.get("provider_server_origin"), "authority provider_server_origin", ) def validated_server_url(authority_origin: str) -> str: raw_server_url = os.environ.get("AGENTSDOCK_SERVER_URL", "").strip() if not raw_server_url: raise JobsCLIError("missing agent environment: AGENTSDOCK_SERVER_URL") server_origin, loopback = canonical_http_origin( raw_server_url, "AGENTSDOCK_SERVER_URL", ) runtime_origin = bounded_identity_value( os.environ.get("AGENTSDOCK_PROVIDER_SERVER_ORIGIN"), "AGENTSDOCK_PROVIDER_SERVER_ORIGIN", ) if runtime_origin: canonical_runtime, _runtime_loopback = canonical_http_origin( runtime_origin, "AGENTSDOCK_PROVIDER_SERVER_ORIGIN", ) if canonical_runtime != server_origin: raise JobsCLIError( "AGENTSDOCK_SERVER_URL conflicts with the live provider origin" ) if loopback: return raw_server_url.rstrip("/") if not authority_origin: raise JobsCLIError( "non-loopback AGENTSDOCK_SERVER_URL must match the authority origin" ) canonical_authority, _authority_loopback = canonical_http_origin( authority_origin, "authority provider_server_origin", ) if canonical_authority != server_origin: raise JobsCLIError( "non-loopback AGENTSDOCK_SERVER_URL must match the authority origin" ) return server_origin def required_environment() -> tuple[str, str, str]: explicit_chat_id = bounded_identity_value( os.environ.get("AGENTSDOCK_CHAT_ID"), "AGENTSDOCK_CHAT_ID", ) token, authority_chat_id = provider_authority() server_url = validated_server_url(authority_server_origin()) if explicit_chat_id and explicit_chat_id != authority_chat_id: raise JobsCLIError("--chat-id does not match the authority file") chat_id = explicit_chat_id or authority_chat_id return server_url, chat_id, token def api_request(method: str, path: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: server_url, _chat_id, token = required_environment() body = json.dumps(payload).encode("utf-8") if payload is not None else None headers = {"Accept": "application/json"} if body is not None: headers["Content-Type"] = "application/json" headers["X-AgentsDock-Provider-Capability"] = token request = urllib.request.Request(f"{server_url}{path}", data=body, headers=headers, method=method) opener = urllib.request.build_opener( urllib.request.ProxyHandler({}), NoRedirectHandler(), ) try: with opener.open(request, timeout=30) as response: decoded = json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") try: detail = json.loads(raw).get("detail") or raw except json.JSONDecodeError: detail = raw raise JobsCLIError(f"server rejected request ({exc.code}): {detail or exc.reason}") from exc except (urllib.error.URLError, TimeoutError, OSError) as exc: reason = getattr(exc, "reason", exc) raise JobsCLIError(f"could not reach AgentsServer: {reason}") from exc except json.JSONDecodeError as exc: raise JobsCLIError("AgentsServer returned invalid JSON") from exc if not isinstance(decoded, dict): raise JobsCLIError("AgentsServer returned an invalid response") return decoded def safe_job_projection(job: dict[str, Any]) -> dict[str, Any]: """Defensively omit exact saved chat references from agent CLI output.""" projected = dict(job) raw_references = projected.pop("chat_references", None) if "chat_target_count" not in projected and isinstance(raw_references, list): projected["chat_target_count"] = len(raw_references) return projected PROVIDER_PRIVATE_CROSS_CHAT_RUN_FIELDS = { "chat_references", "target_session_id", "source_session_id", "source_run_id", "authorization_source_run_id", "authorization_route_id", "cross_chat_envelope_id", "cross_chat_exchange_id", "cross_chat_exchange_leg_id", "exchange_id", "exchange_leg_id", "inbound_leg_id", "envelope_id", "leg_id", "parent_leg_id", "cross_chat_obligation_ids", "cross_chat_exchange_ids", "cross_chat_direct_message_ids", "secure_peer_envelope_id", "requester_session_id", "responder_session_id", "target_chat_id", } def contains_private_chat_target_fields(value: Any) -> bool: if isinstance(value, dict): return any( key in PROVIDER_PRIVATE_CROSS_CHAT_RUN_FIELDS or ( key == "id" and isinstance(nested, str) and re.fullmatch( r"(?:handoff|exchange|leg)_[A-Za-z0-9_-]+", nested, ) ) or contains_private_chat_target_fields(nested) for key, nested in value.items() ) if isinstance(value, list): return any(contains_private_chat_target_fields(nested) for nested in value) return False def scoped_jobs() -> list[dict[str, Any]]: _server_url, chat_id, _token = required_environment() encoded_chat_id = urllib.parse.quote(chat_id, safe="") response = api_request("GET", f"/api/agent/sessions/{encoded_chat_id}/jobs") jobs = response.get("jobs") if not isinstance(jobs, list) or not all(isinstance(job, dict) for job in jobs): raise JobsCLIError("AgentsServer returned an invalid jobs list") foreign = [str(job.get("id") or "unknown") for job in jobs if job.get("session_id") != chat_id] if foreign: raise JobsCLIError("AgentsServer returned jobs outside the active chat scope") return [safe_job_projection(job) for job in jobs] def owned_job(job_id: str) -> dict[str, Any]: for job in scoped_jobs(): if job.get("id") == job_id: return job raise JobsCLIError(f"job {job_id!r} does not exist in the active chat") def checked_job(response: dict[str, Any]) -> dict[str, Any]: _server_url, chat_id, _token = required_environment() job = response.get("job") if not isinstance(job, dict): raise JobsCLIError("AgentsServer returned an invalid job") if job.get("session_id") != chat_id: raise JobsCLIError("AgentsServer returned a job outside the active chat scope") return safe_job_projection(job) def command_list(_args: argparse.Namespace) -> Any: return {"jobs": scoped_jobs()} def command_get(args: argparse.Namespace) -> Any: return {"job": owned_job(args.job_id)} def command_runs(args: argparse.Namespace) -> Any: _server_url, chat_id, _token = required_environment() owned_job(args.job_id) encoded_chat_id = urllib.parse.quote(chat_id, safe="") job_id = urllib.parse.quote(args.job_id, safe="") query = urllib.parse.urlencode({ key: value for key, value in ( ("before_seq", args.before_seq), ("limit", args.limit), ) if value is not None }) path = f"/api/agent/sessions/{encoded_chat_id}/jobs/{job_id}/runs" if query: path += f"?{query}" response = api_request("GET", path) runs = response.get("runs") if not isinstance(runs, list) or not all( isinstance(run, dict) for run in runs ): raise JobsCLIError("AgentsServer returned invalid job history") if ( response.get("session_id") != chat_id or response.get("job_id") != args.job_id ): raise JobsCLIError("AgentsServer returned history outside the active chat scope") if contains_private_chat_target_fields(response): raise JobsCLIError("AgentsServer returned private chat target data in job history") return response def command_create(args: argparse.Namespace) -> Any: _server_url, chat_id, _token = required_environment() if args.interval_seconds is None and args.cron is None and args.rrule is None and args.first_run_at is None: raise JobsCLIError("create requires --interval-seconds, --cron, --rrule, or --first-run-at") if args.interval_seconds is None and args.cron is None and args.rrule is None: if args.loop or (args.max_runs is not None and args.max_runs != 1): raise JobsCLIError("a one-time --first-run-at job cannot loop or run more than once without a schedule") schedule_kind = "cron" if args.cron is not None else "rrule" if args.rrule is not None else "interval" payload: dict[str, Any] = { "title": args.title, "prompt": args.prompt, "schedule_kind": schedule_kind, "interval_seconds": args.interval_seconds, "cron_expression": args.cron, "rrule": args.rrule, "timezone": args.timezone, "first_run_at": args.first_run_at, "loop": args.loop, "max_runs": args.max_runs, "enabled": not args.disabled, "backend": args.backend, "context_mode": args.context_mode, "chat_routes": list(args.chat_route or []), } encoded_chat_id = urllib.parse.quote(chat_id, safe="") return {"job": checked_job(api_request("POST", f"/api/agent/sessions/{encoded_chat_id}/jobs", payload))} def command_update(args: argparse.Namespace) -> Any: _server_url, chat_id, _token = required_environment() current_job = owned_job(args.job_id) patch: dict[str, Any] = {} for key in ( "title", "prompt", "interval_seconds", "rrule", "timezone", "next_run_at", "backend", "context_mode", ): value = getattr(args, key) if value is not None: patch[key] = value target_kind = ( "interval" if args.interval_seconds is not None else "cron" if args.cron is not None else "rrule" if args.rrule is not None else str(current_job.get("schedule_kind") or "interval") ) if args.max_runs is not None: patch["max_runs"] = args.max_runs if target_kind == "interval": patch["loop"] = False elif args.unlimited: patch["max_runs"] = None if target_kind == "interval": patch["loop"] = True if args.cron is not None: patch["cron_expression"] = args.cron if args.interval_seconds is not None: patch["schedule_kind"] = "interval" elif args.cron is not None: patch["schedule_kind"] = "cron" elif args.rrule is not None: patch["schedule_kind"] = "rrule" if args.loop is not None: patch["loop"] = args.loop if args.enabled is not None: patch["enabled"] = args.enabled if args.chat_route is not None: patch["chat_routes"] = list(args.chat_route) elif args.clear_chat_routes: patch["chat_routes"] = [] if not patch: raise JobsCLIError("update requires at least one changed field") encoded_chat_id = urllib.parse.quote(chat_id, safe="") job_id = urllib.parse.quote(args.job_id, safe="") return { "job": checked_job( api_request("PATCH", f"/api/agent/sessions/{encoded_chat_id}/jobs/{job_id}", patch) ) } def command_delete(args: argparse.Namespace) -> Any: _server_url, chat_id, _token = required_environment() owned_job(args.job_id) encoded_chat_id = urllib.parse.quote(chat_id, safe="") job_id = urllib.parse.quote(args.job_id, safe="") response = api_request("DELETE", f"/api/agent/sessions/{encoded_chat_id}/jobs/{job_id}") if response.get("deleted") is not True: raise JobsCLIError(f"job {args.job_id!r} was not deleted") return {"ok": True, "deleted": True, "job_id": args.job_id} def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Manage scheduled jobs for the current AgentsDock chat.", allow_abbrev=False, ) parser.add_argument( "--authority-file", help="mode-0600 per-run AgentsDock provider authority file", ) parser.add_argument( "--chat-id", type=nonempty_chat_id, help="explicit chat scope (defaults to AGENTSDOCK_CHAT_ID)", ) subparsers = parser.add_subparsers(dest="command", required=True) list_parser = subparsers.add_parser( "list", help="list jobs in the active chat", allow_abbrev=False ) list_parser.set_defaults(handler=command_list) get_parser = subparsers.add_parser( "get", help="get one job in the active chat", allow_abbrev=False ) get_parser.add_argument("job_id") get_parser.set_defaults(handler=command_get) runs_parser = subparsers.add_parser( "runs", help="show recent run status for one job", allow_abbrev=False ) runs_parser.add_argument("job_id") runs_parser.add_argument("--before-seq", type=int) runs_parser.add_argument("--limit", type=int, default=20) runs_parser.set_defaults(handler=command_runs) create_parser = subparsers.add_parser( "create", help="create a job in the active chat", allow_abbrev=False ) create_parser.add_argument("--title", required=True) create_parser.add_argument("--prompt", required=True) create_schedule = create_parser.add_mutually_exclusive_group() create_schedule.add_argument("--interval-seconds", type=int) create_schedule.add_argument("--cron", help="cron expression (seconds-first for 6-7 fields)") create_schedule.add_argument("--rrule", help="RFC 5545 RRULE property") create_parser.add_argument("--timezone", default="UTC", help="IANA timezone (default: UTC)") create_parser.add_argument("--first-run-at", help="ISO-8601 timestamp for the first run") create_parser.add_argument("--loop", action="store_true", help="repeat at the interval") create_parser.add_argument("--max-runs", type=int) create_parser.add_argument("--disabled", action="store_true") create_parser.add_argument("--backend", choices=("codex", "claude", "cursor")) create_parser.add_argument( "--context-mode", choices=("chat", "standalone"), default="chat", help="continue in the parent chat (default) or use a fresh provider context", ) create_parser.add_argument( "--chat-route", action="append", type=chat_route_selection, metavar="ROUTE_ID[=ACTION]", help=( "persist an exact chat target from the current chats-helper list; " "the prompt must contain that chat's exact @title (repeatable)" ), ) create_parser.set_defaults(handler=command_create) update_parser = subparsers.add_parser( "update", help="update a job owned by the active chat", allow_abbrev=False ) update_parser.add_argument("job_id") update_parser.add_argument("--title") update_parser.add_argument("--prompt") update_schedule = update_parser.add_mutually_exclusive_group() update_schedule.add_argument("--interval-seconds", type=int) update_schedule.add_argument("--cron", help="cron expression (seconds-first for 6-7 fields)") update_schedule.add_argument("--rrule", help="RFC 5545 RRULE property") update_parser.add_argument("--timezone", help="IANA timezone") update_parser.add_argument("--next-run-at", help="ISO-8601 timestamp for the next run") run_limit_group = update_parser.add_mutually_exclusive_group() run_limit_group.add_argument("--max-runs", type=int) run_limit_group.add_argument("--unlimited", action="store_true", help="clear a finite run limit") update_parser.add_argument("--backend", choices=("codex", "claude", "cursor")) update_parser.add_argument( "--context-mode", choices=("chat", "standalone"), ) loop_group = update_parser.add_mutually_exclusive_group() loop_group.add_argument("--loop", dest="loop", action="store_true") loop_group.add_argument("--no-loop", dest="loop", action="store_false") enabled_group = update_parser.add_mutually_exclusive_group() enabled_group.add_argument("--enable", dest="enabled", action="store_true") enabled_group.add_argument("--disable", dest="enabled", action="store_false") update_routes = update_parser.add_mutually_exclusive_group() update_routes.add_argument( "--chat-route", action="append", type=chat_route_selection, metavar="ROUTE_ID[=ACTION]", help=( "replace persisted chat targets using opaque routes from the " "current chats-helper list (repeatable)" ), ) update_routes.add_argument( "--clear-chat-routes", action="store_true", help="revoke every persisted chat target from this job", ) update_parser.set_defaults( handler=command_update, loop=None, enabled=None, chat_route=None, clear_chat_routes=False, ) delete_parser = subparsers.add_parser( "delete", help="delete a job owned by the active chat", allow_abbrev=False ) delete_parser.add_argument("job_id") delete_parser.set_defaults(handler=command_delete) return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) previous_authority_file = os.environ.get( "AGENTSDOCK_PROVIDER_AUTHORITY_FILE" ) previous_chat_id = os.environ.get("AGENTSDOCK_CHAT_ID") try: try: authority_path = selected_authority_path(args.authority_file) _token, authority_chat_id = provider_authority(args.authority_file) explicit_chat_id = bounded_identity_value(args.chat_id, "--chat-id") environment_chat_id = bounded_identity_value( previous_chat_id, "AGENTSDOCK_CHAT_ID", ) if ( explicit_chat_id and environment_chat_id and explicit_chat_id != environment_chat_id ): raise JobsCLIError( "--chat-id conflicts with AGENTSDOCK_CHAT_ID" ) for candidate in (explicit_chat_id, environment_chat_id): if candidate and candidate != authority_chat_id: raise JobsCLIError( "--chat-id does not match the authority file" ) os.environ["AGENTSDOCK_PROVIDER_AUTHORITY_FILE"] = str( authority_path ) os.environ["AGENTSDOCK_CHAT_ID"] = authority_chat_id result = args.handler(args) except JobsCLIError as exc: print(f"error: {exc}", file=sys.stderr) return 1 finally: if previous_authority_file is None: os.environ.pop("AGENTSDOCK_PROVIDER_AUTHORITY_FILE", None) else: os.environ[ "AGENTSDOCK_PROVIDER_AUTHORITY_FILE" ] = previous_authority_file if previous_chat_id is None: os.environ.pop("AGENTSDOCK_CHAT_ID", None) else: os.environ["AGENTSDOCK_CHAT_ID"] = previous_chat_id print(json.dumps(result, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())