AgentContextWriter

action batch 1  →  write_actions  →  destination (raw, latest context)
action batch 2  →  write_actions  →  destination (raw, latest context)
...K actions    →  AsyncSessionSummarizer  →  LLM
                →  overwrite_with_summary  →  destination (condensed summary replaces raw actions)
action batch K+1  →  write_actions  →  destination (fresh raw context, cycle restarts)

import asyncio
import httpx
import openai
from collections import defaultdict
from autoplay_sdk import AsyncConnectorClient, AsyncSessionSummarizer
from autoplay_sdk.agent_context import AsyncAgentContextWriter
from autoplay_sdk.models import ActionsPayload

async_openai = openai.AsyncOpenAI()

# ---------------------------------------------------------------------------
# Intercom HTTP client (wraps _post_note and _redact_part)
# ---------------------------------------------------------------------------

INTERCOM_API = "https://api.intercom.io"
ACCESS_TOKEN = "your-intercom-access-token"
ADMIN_ID = "your-admin-id"

http = httpx.AsyncClient(
    base_url=INTERCOM_API,
    headers={"Authorization": f"Bearer {ACCESS_TOKEN}", "Accept": "application/json"},
)

async def post_note(conversation_id: str, body: str) -> str | None:
    """Post an internal admin note and return the part_id on success."""
    r = await http.post(
        f"/conversations/{conversation_id}/reply",
        json={"type": "admin", "admin_id": ADMIN_ID, "message_type": "note", "body": body},
    )
    if r.is_success:
        return r.json().get("id")
    return None

async def redact_part(conversation_id: str, part_id: str) -> None:
    """Blank a previously posted note (best-effort — failures are swallowed)."""
    await http.post(
        "/conversations/redact",
        json={"type": "conversation_part", "conversation_id": conversation_id, "conversation_part_id": part_id},
    )

# ---------------------------------------------------------------------------
# Closure state (one instance per product / worker)
# ---------------------------------------------------------------------------

# session_id → Intercom conversation_id (populated when a conversation opens)
conv_map: dict[str, str] = {}
# session_id → list of part_ids for raw action notes (tracked for later redaction)
part_ids: dict[str, list[str]] = defaultdict(list)

# ---------------------------------------------------------------------------
# Callbacks
# ---------------------------------------------------------------------------

async def write_actions_cb(session_id: str, text: str) -> None:
    conv_id = conv_map.get(session_id)
    if not conv_id:
        return
    part_id = await post_note(conv_id, text)
    if part_id:
        part_ids[session_id].append(part_id)

async def overwrite_cb(session_id: str, summary: str) -> None:
    conv_id = conv_map.get(session_id)
    if not conv_id:
        return
    # Summary posted first — agent never has a blank context window.
    await post_note(conv_id, summary)
    old = part_ids.pop(session_id, [])
    if old:
        # Redact all old action notes in parallel.
        await asyncio.gather(*[redact_part(conv_id, pid) for pid in old])

# ---------------------------------------------------------------------------
# LLM — wrap with a circuit breaker in production to avoid flooding a
# degraded endpoint. Here shown as a plain async call for clarity.
# ---------------------------------------------------------------------------

async def llm(prompt: str) -> str:
    r = await async_openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
        max_tokens=256,
    )
    return r.choices[0].message.content

# ---------------------------------------------------------------------------
# Custom action formatter — groups actions by page for a cleaner LLM prompt.
# Remove format_actions to use the default numbered list instead.
# ---------------------------------------------------------------------------

def format_actions(payloads: list[ActionsPayload]) -> str:
    lines = []
    current_url = None
    for p in payloads:
        for a in p.actions:
            if a.canonical_url != current_url:
                current_url = a.canonical_url
                lines.append(f"\n[{current_url}]")
            lines.append(f"  - {a.title}")
    return "\n".join(lines).strip()

# ---------------------------------------------------------------------------
# Wire it together
# ---------------------------------------------------------------------------

summarizer = AsyncSessionSummarizer(
    llm=llm,
    threshold=10,          # summarise every 10 actions
    format_actions=format_actions,
)

writer = AsyncAgentContextWriter(
    summarizer=summarizer,
    write_actions=write_actions_cb,
    overwrite_with_summary=overwrite_cb,
)

async with AsyncConnectorClient(url=CONNECTOR_URL, token=API_TOKEN) as client:
    client.on_actions(writer.add)
    await client.run()

from autoplay_sdk import AsyncConnectorClient, AsyncSessionSummarizer
from autoplay_sdk.agent_context import AsyncAgentContextWriter

summarizer = AsyncSessionSummarizer(llm=my_llm, threshold=10)

async def write_actions(session_id: str, text: str) -> None:
    # Keep the chatbot's context window current between summarisations.
    await chatbot.set_context(session_id, text)

async def overwrite_with_summary(session_id: str, summary: str) -> None:
    # Replace accumulated raw actions with the compressed summary.
    await chatbot.set_context(session_id, summary)

writer = AsyncAgentContextWriter(
    summarizer=summarizer,
    overwrite_with_summary=overwrite_with_summary,
    write_actions=write_actions,  # optional — omit if you only want summaries pushed
)

async with AsyncConnectorClient(url=URL, token=TOKEN) as client:
    client.on_actions(writer.add)
    await client.run()

# Reduces write_actions calls when PostHog sends event bursts
writer = AsyncAgentContextWriter(
    summarizer=summarizer,
    write_actions=my_destination_cb,
    overwrite_with_summary=overwrite_cb,
    debounce_ms=200,  # wait 200ms of silence before dispatching
)