Developer API

Immutable rules (send once in session instructions):
- Never press Escape or close the Group Note window.
- Paste only with one Ctrl+V chord after focus is visible.

Micro-goal A (send now):
Goal: focus the Group Note body.
Expected screen: the caret or focus highlight is visible in the note body.
[done] when: note-body focus is visible.

Micro-goal B (send only after A is verified):
Goal: paste the clipboard once with one Ctrl+V chord.
Expected screen: the note body visibly contains the pasted text.
[done] when: the pasted text is visible in the note body.

import base64, email.utils, os, random, time, uuid, requests
from datetime import timezone

BASE = "https://coasty.ai/v1"
HEADERS = {"X-API-Key": os.environ["COASTY_API_KEY"]}
MICRO_GOAL = "Open tomorrow at 3:00 PM. Expected screen: the day view shows 3:00 PM. [done] when that slot is visible."
VERIFY = "Verify only whether tomorrow's 3:00 PM slot is visible. Return done if visible; otherwise describe the mismatch. Do not click or type."

def screenshot() -> str:
    # Replace this with a capture made NOW; never reuse a pre-action image.
    with open("screen.png", "rb") as f:
        return base64.b64encode(f.read()).decode()

def prohibited(action) -> bool:
    params = action.get("params", {})
    values = []
    for field in ("key", "keys", "hold_keys", "modifiers"):
        raw = params.get(field)
        values.extend([raw] if isinstance(raw, str) else (raw or []))
    aliases = {"esc": "escape", "control": "ctrl", "ctl": "ctrl", "command": "cmd", "super": "cmd", "win": "cmd", "windows": "cmd", "meta": "cmd", "option": "alt", "ctrl_l": "ctrl", "ctrl_r": "ctrl", "control_l": "ctrl", "control_r": "ctrl", "cmd_l": "cmd", "cmd_r": "cmd", "command_l": "cmd", "command_r": "cmd", "meta_l": "cmd", "meta_r": "cmd", "super_l": "cmd", "super_r": "cmd", "win_l": "cmd", "win_r": "cmd", "windows_l": "cmd", "windows_r": "cmd", "alt_l": "alt", "alt_r": "alt", "option_l": "alt", "option_r": "alt"}
    keys = set()
    for value in values:
        for part in str(value).strip().lower().replace("-", "+").split("+"):
            part = part.strip()
            if part:
                keys.add(aliases.get(part, part))
    close_actions = {"close_window", "window_close", "browser_close", "browser_close_tab", "close_tab", "terminal_close"}
    return (
        "escape" in keys or action["action_type"] in close_actions
        or {"alt", "f4"} <= keys
        or (("ctrl" in keys or "cmd" in keys) and "w" in keys)
        or {"cmd", "q"} <= keys
    )

def policy_failure(result) -> bool:
    return any(
        a["action_type"] == "fail"
        and a.get("params", {}).get("code") == "ACTION_POLICY_VIOLATION"
        for a in result["actions"]
    )

def parse_retry_after_seconds(raw):
    if raw is None:
        return None
    try:
        return max(0.0, float(raw))
    except (TypeError, ValueError):
        try:
            value = email.utils.parsedate_to_datetime(str(raw))
            if value.tzinfo is None:
                value = value.replace(tzinfo=timezone.utc)
            return max(0.0, value.timestamp() - time.time())
        except (TypeError, ValueError, OverflowError):
            return None

def cleanup_retry_after_seconds(response):
    try:
        envelope = response.json()
        if not isinstance(envelope, dict):
            envelope = {}
        error = envelope.get("error") or {}
        if not isinstance(error, dict):
            error = {}
    except (ValueError, AttributeError, requests.RequestException):
        error = {}
    values = [
        parse_retry_after_seconds(response.headers.get("Retry-After")),
        parse_retry_after_seconds(error.get("retry_after")),
    ]
    valid = [value for value in values if value is not None]
    return max(valid) if valid else None

def close_session(session_id: str) -> None:
    url = f"{BASE}/sessions/{session_id}"
    deadline = time.monotonic() + 120
    for attempt in range(3):
        server_delay = None
        try:
            deleted = requests.delete(url, headers=HEADERS, timeout=15)
        except requests.RequestException:
            deleted = None
        if deleted is not None:
            if deleted.ok or deleted.status_code == 404:
                return
            if deleted.status_code not in (429, 502, 503, 504):
                deleted.raise_for_status()
            server_delay = cleanup_retry_after_seconds(deleted)
        # A lost DELETE response is ambiguous. GET is authoritative.
        try:
            state = requests.get(url, headers=HEADERS, timeout=15)
            if state.status_code == 404:
                return
        except requests.RequestException:
            pass
        if attempt < 2:
            delay = max(server_delay or 0, random.uniform(0, min(4, 2 ** attempt)))
            if delay > deadline - time.monotonic():
                raise RuntimeError(f"cleanup retry exceeds budget; honor Retry-After and resume GET {url}")
            time.sleep(delay)
    raise RuntimeError(f"session cleanup unconfirmed; reconcile GET {url}")

# 1. Open a session — it remembers the trajectory across steps
session_payload = {
    "screen_width": 1920,
    "screen_height": 1080,
    "action_policy": {
        "blocked_keys": ["escape"],
        "block_window_close": True,
        "max_actions": 5,
    },
}
create_key = f"session-create-{uuid.uuid4().hex}"
# If transport fails, retry only this same key + exact payload. Reconcile an
# ambiguous result at GET /v1/idempotency/{create_key} before using a new key.
created = requests.post(
    f"{BASE}/sessions",
    headers={**HEADERS, "Idempotency-Key": create_key},
    json=session_payload,
    timeout=60,
)
created.raise_for_status()
session = created.json()
session_id = session["session_id"]

# 2. Drive the task one step at a time
instruction = MICRO_GOAL
completed = False
rejections = 0
try:
    for step in range(20):  # safety cap
        response = requests.post(
            f"{BASE}/sessions/{session_id}/predict",
            headers={**HEADERS, "Idempotency-Key": f"step-{session_id}-{step}-{uuid.uuid4().hex}"},
            json={"screenshot": screenshot(), "instruction": instruction},
            timeout=60,
        )
        response.raise_for_status()
        res = response.json()

        # Server policy replaces a prohibited batch with one fail action. The
        # local scan is defense in depth. Execute none of that batch.
        if policy_failure(res) or any(prohibited(a) for a in res["actions"]):
            rejections += 1
            if rejections >= 3:
                raise RuntimeError("three prohibited proposals; human review required")
            instruction = "CONTROL FEEDBACK: the previous prohibited action was rejected and NOT executed. The next screenshot is authoritative. Continue without Escape."
            continue
        rejections = 0
        if res["status"] == "fail":
            raise RuntimeError(res.get("reasoning", "agent failed"))

        for action in res["actions"]:
            if action["action_type"] not in ("done", "fail"):
                perform(action)      # your guarded action executor
        time.sleep(0.5)              # replace with a bounded DOM/app-state wait when available

        if res["status"] == "done":
            # done is a model claim: observe AFTER the actions, then verify.
            verify_response = requests.post(
                f"{BASE}/sessions/{session_id}/predict",
                headers={**HEADERS, "Idempotency-Key": f"verify-{session_id}-{uuid.uuid4().hex}"},
                json={"screenshot": screenshot(), "instruction": VERIFY},
                timeout=60,
            )
            verify_response.raise_for_status()
            verified = verify_response.json()
            if verified["status"] != "done" or any(
                a["action_type"] not in ("done", "fail") for a in verified["actions"]
            ):
                raise RuntimeError("completion did not match the fresh screenshot")
            completed = True
            break
        instruction = MICRO_GOAL
    if not completed:
        raise RuntimeError("step cap reached before verified completion")
finally:
    # 3. Always release the session to free your concurrency quota
    close_session(session_id)