AI Agent Lifecycle Hooks: Run Code on Agent Events

from promptise.runtime import HookBlocked, HookConfig, ProcessConfig

async def no_passwords(ctx):
    if "password" in ctx.data["prompt"].lower():
        raise HookBlocked("password tickets go to the identity team")

config = ProcessConfig(
    instructions="You triage support tickets.",
    hooks=[HookConfig(event="user_prompt_submit", callback=no_passwords, fail_closed=True)],
)

start() ─▶ SESSION_START
              │
trigger or inject() ─▶ [FILE_CHANGED*] ─▶ TASK_CREATED* ─▶ queue
                                                            │
                                     worker takes the event ▼
                          USER_PROMPT_SUBMIT* (read or rewrite the prompt)
                                                            │
                                   the agent run ───────────┤
                           tool needs approval ─▶ PERMISSION_REQUEST*
                     approval, budget or guardrail refused ─▶ PERMISSION_DENIED
                         open mode spawns a process ─▶ SUBAGENT_START*  (SUBAGENT_STOP later)
              the agent or an operator changes config ─▶ CONFIG_CHANGE* (* for the agent's changes)
                                                            │
                     history over its limit ─▶ PRE_COMPACT ─▶ POST_COMPACT
                                                            ▼
                                                     TASK_COMPLETED
stop() or restart ─▶ SESSION_END

pip install "promptise>=1.5.0"

"""Step 1: register one hook for all 13 events and watch which ones a process fires."""
import asyncio
import os
import time

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8781/v1"  # the fake model server
os.environ["OPENAI_API_KEY"] = "not-a-real-key"

from fake_llm import FakeLLM
from promptise.runtime import (
    AgentProcess, HookConfig, HookContext, HookEvent, ProcessConfig, TriggerEvent,
)

T0 = time.perf_counter()
SHOWN = ("restart", "recovered", "reason", "trigger_type", "prompt", "outcome", "duration_ms")


async def trace(ctx: HookContext) -> None:
    data = {k: v for k, v in ctx.data.items() if k in SHOWN}
    if "prompt" in data:
        data["prompt"] = data["prompt"][:30] + "..."
    print(f"{time.perf_counter() - T0:5.2f}s  {ctx.event.value:<19} {data}")


config = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions="You triage support tickets.",
    hooks=[HookConfig(event=event.value, callback=trace) for event in HookEvent],
)


async def main() -> None:
    llm = FakeLLM()
    await llm.start()
    process = AgentProcess("triage", config)
    print("hooks registered:", process.status()["hook_count"])
    await process.start()
    await process.inject(TriggerEvent(
        trigger_id="manual", trigger_type="manual", payload={"ticket": "Login page is down"},
    ))
    await asyncio.sleep(1.0)
    await process.stop()
    await llm.stop()


asyncio.run(main())

"""Step 2: refuse events and runs, rewrite the prompt, and see fail_closed at work."""
import asyncio
import logging
import os

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8781/v1"
os.environ["OPENAI_API_KEY"] = "not-a-real-key"
logging.basicConfig(level=logging.ERROR)  # keep the hook-error tracebacks out of the output

from fake_llm import FakeLLM
from promptise.runtime import (
    AgentProcess, HookBlocked, HookConfig, HookContext, HookEvent, HookManager,
    ProcessConfig, TriggerEvent,
)


async def retired_triggers(ctx: HookContext) -> None:
    # task_created: before the event is queued. Data has no payload, only IDs.
    if ctx.data["trigger_id"] == "legacy-cron":
        raise HookBlocked("legacy-cron was retired")


async def no_credentials(ctx: HookContext) -> None:
    # user_prompt_submit: the full prompt, payload included.
    if "password" in ctx.data["prompt"].lower():
        raise HookBlocked("tickets about passwords go to the identity team")


async def house_rules(ctx: HookContext) -> None:
    ctx.set("prompt", ctx.data["prompt"] + "\n\nHouse rules: answer in one sentence.")


config = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions="You triage support tickets.",
    hooks=[
        HookConfig(event="task_created", callback=retired_triggers),
        HookConfig(event="user_prompt_submit", callback=no_credentials,
                   priority=10, fail_closed=True),
        HookConfig(event="user_prompt_submit", callback=house_rules),
    ],
)


def ticket(text: str, trigger_id: str = "helpdesk") -> TriggerEvent:
    return TriggerEvent(trigger_id=trigger_id, trigger_type="webhook", payload={"ticket": text})


async def main() -> None:
    llm = FakeLLM()
    await llm.start()
    process = AgentProcess("triage", config)
    await process.start()
    await process.inject(ticket("Nightly export failed", trigger_id="legacy-cron"))
    await process.inject(ticket("Please reset my password"))
    await process.inject(ticket("Login page is down"))
    await asyncio.sleep(1.5)

    print("runs:", process.status()["invocation_count"])
    for dead in process.dead_letters:
        print("dead-lettered:", dead["event"].trigger_id, "|", dead["reason"], "|", dead["error"])
    print("the model received:", "..." + llm.prompts[-1][-75:].replace("\n", " "))
    await process.stop()
    await llm.stop()

    # fail_closed decides what a broken or slow hook means.
    async def slow_policy_service(ctx: HookContext) -> None:
        await asyncio.sleep(2)

    for fail_closed in (False, True):
        hooks = HookManager()
        hooks.register(HookEvent.USER_PROMPT_SUBMIT, slow_policy_service,
                       timeout=0.5, fail_closed=fail_closed)
        result = await hooks.dispatch(HookEvent.USER_PROMPT_SUBMIT, data={"prompt": "hi"})
        print(f"fail_closed={fail_closed}: blocked={result.blocked} "
              f"errors={len(result.errors)} reason={result.reason!r}")


asyncio.run(main())

#!/usr/bin/env python3
"""A command hook: the event arrives as JSON on stdin, the answer goes to stdout."""
import json
import os
import sys

event = json.load(sys.stdin)
prompt = event["data"]["prompt"]

with open(os.environ["AUDIT_LOG"], "a") as log:  # AUDIT_LOG comes from the hook's env
    log.write(json.dumps({
        "event": os.environ["PROMPTISE_HOOK_EVENT"],
        "process": os.environ["PROMPTISE_PROCESS"],
        "event_id": event["data"]["event_id"],
        "queue": os.environ.get("TICKET_QUEUE"),          # passed through by name
        "sees_api_key": "OPENAI_API_KEY" in os.environ,   # not passed
    }) + "\n")

if "refund" in prompt.lower():
    print(json.dumps({"block": True, "reason": "refunds need a human"}))
# No output: allow the prompt unchanged.

"""Step 3: command hooks (ShellHook): stdin JSON, stdout JSON, a minimal environment."""
import asyncio
import json
import logging
import os
import sys
from pathlib import Path

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8781/v1"
os.environ["OPENAI_API_KEY"] = "not-a-real-key"
os.environ["TICKET_QUEUE"] = "support-eu"
logging.basicConfig(level=logging.CRITICAL)

from fake_llm import FakeLLM
from promptise.runtime import (
    AgentProcess, HookConfig, HookContext, HookEvent, ProcessConfig, ShellHook, TriggerEvent,
)

HERE = Path(__file__).parent
AUDIT_LOG = HERE / "out" / "audit.log"

config = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions="You triage support tickets.",
    hooks=[
        HookConfig(
            event="user_prompt_submit",
            command=[sys.executable, str(HERE / "hooks" / "gate.py")],
            timeout=5,
            fail_closed=True,
            env={"AUDIT_LOG": str(AUDIT_LOG)},
            env_passthrough=["TICKET_QUEUE"],
        ),
    ],
)


async def main() -> None:
    AUDIT_LOG.unlink(missing_ok=True)
    llm = FakeLLM()
    await llm.start()
    process = AgentProcess("triage", config)
    await process.start()
    for text in ("Please refund order 1182", "Printer on floor 3 is jammed"):
        await process.inject(TriggerEvent(trigger_id="helpdesk", trigger_type="webhook",
                                          payload={"ticket": text}))
    await asyncio.sleep(2)
    print("runs:", process.status()["invocation_count"])
    for dead in process.dead_letters:
        print("dead-lettered:", dead["reason"], "|", dead["error"])
    for line in AUDIT_LOG.read_text().splitlines():
        print("audit:", line)
    await process.stop()
    await llm.stop()

    # A non-zero exit is a broken hook, not a block.
    broken = ShellHook(command=[sys.executable, str(HERE / "hooks" / "broken.py")])
    ctx = HookContext(event=HookEvent.TASK_COMPLETED, process_id="triage", data={})
    try:
        await broken.run(ctx)
    except RuntimeError as exc:
        print("exit 2 ->", type(exc).__name__ + ":",
              str(exc).replace(sys.executable, "python").replace(f"{HERE}/", ""))


asyncio.run(main())

version: "1.0"
name: triage
model: openai:gpt-5-mini
instructions: You triage support tickets.
triggers:
  - type: webhook
    webhook_path: /tickets
    webhook_port: 9782
hooks:
  - event: user_prompt_submit
    command: ["python3", "hooks/gate.py"]
    timeout: 5
    fail_closed: true
    env_passthrough: [TICKET_QUEUE, AUDIT_LOG]
  - event: task_completed
    command: "python3 hooks/notify.py --channel ops"

"""Step 4: hooks in a manifest, and why the control API refuses to deploy them."""
import asyncio
import logging

import httpx

from promptise.runtime import (
    AgentRuntime, OrchestrationAPI, load_manifest, manifest_to_process_config,
)

logging.basicConfig(level=logging.CRITICAL)

config = manifest_to_process_config(load_manifest("triage.agent"))
for hook in config.hooks:
    print(f"{hook.event:<19} command={hook.command} timeout={hook.timeout} "
          f"fail_closed={hook.fail_closed}")

DEPLOY = {
    "name": "triage",
    "config": {
        "instructions": "You triage support tickets.",
        "hooks": [{"event": "user_prompt_submit", "command": ["python3", "hooks/gate.py"]}],
    },
}


async def deploy(allow_command_deploys: bool) -> None:
    runtime = AgentRuntime()
    api = OrchestrationAPI(runtime, port=9781, allow_command_deploys=allow_command_deploys)
    await api.start()
    async with httpx.AsyncClient() as client:
        r = await client.post("http://127.0.0.1:9781/api/v1/processes", json=DEPLOY)
    body = r.json()
    print(f"allow_command_deploys={allow_command_deploys}: HTTP {r.status_code}",
          body.get("error", {}).get("code", body.get("name")))
    if r.status_code == 403:
        print("  ", body["error"]["message"][:118] + "...")
    await api.stop()
    await runtime.stop_all()


async def main() -> None:
    await deploy(allow_command_deploys=False)
    await deploy(allow_command_deploys=True)


asyncio.run(main())

"""Step 5: gate tool approvals with permission_request, audit refusals with a shared manager."""
import asyncio
import logging
import os

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8781/v1"
os.environ["OPENAI_API_KEY"] = "not-a-real-key"
logging.basicConfig(level=logging.CRITICAL)

from fake_llm import FakeLLM
from promptise.approval import ApprovalDecision, ApprovalPolicy
from promptise.runtime import (
    AgentRuntime, ContextConfig, HookBlocked, HookConfig, HookContext, HookEvent,
    HookManager, ProcessConfig, TriggerEvent,
)

asked: list[str] = []


async def reviewer(request) -> ApprovalDecision:  # stands in for Slack, a queue, a UI
    asked.append(f"{request.tool_name}({request.arguments})")
    return ApprovalDecision(approved=True, reviewer_id="alice")


async def no_refund_limits(ctx: HookContext) -> None:
    if ctx.data["arguments"].get("key") == "refund_limit":
        raise HookBlocked("refund limits are set by finance, not by the agent")


# One manager for every process in the runtime.
shared = HookManager()


@shared.on(HookEvent.PERMISSION_DENIED)
async def audit_denials(ctx: HookContext) -> None:
    d = ctx.data
    print(f"denied in {ctx.process_id}: kind={d['kind']} tool={d['tool_name']} "
          f"decided_by={d['decided_by']} reason={d['reason']!r}")


config = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions="You triage support tickets and record decisions with update_state.",
    context=ContextConfig(state_tool=True),
    approval=ApprovalPolicy(tools=["update_state"], handler=reviewer),
    hooks=[HookConfig(event="permission_request", callback=no_refund_limits, fail_closed=True)],
)


async def main() -> None:
    llm = FakeLLM()
    llm.tool_calls = [
        ("update_state", {"key": "priority", "value": "high"}),
        ("update_state", {"key": "refund_limit", "value": 5000}),
    ]
    await llm.start()
    runtime = AgentRuntime(hooks=shared)
    process = await runtime.add_process("triage", config)
    await runtime.start_all()
    for text in ("Checkout fails for all EU customers", "Customer wants a 5000 EUR refund"):
        await process.inject(TriggerEvent(trigger_id="helpdesk", trigger_type="webhook",
                                          payload={"ticket": text}))
    await asyncio.sleep(2)
    print("reviewer was asked:", asked)
    print("state:", process.context.state_snapshot())
    await runtime.stop_all()
    await llm.stop()


asyncio.run(main())

lifecycle-hooks/
├── fake_llm.py              # scripted OpenAI-compatible model server
├── hooks_tour.py            # step 1: one hook on all 13 events
├── policy_hooks.py          # step 2: block, rewrite, fail_closed
├── hooks/gate.py            # step 3: a command hook (stdin/stdout JSON)
├── hooks/broken.py          # step 3: a command hook that exits 2
├── shell_hooks.py           # step 3: ShellHook environment and contract
├── triage.agent             # step 4: hooks in a manifest
├── manifest_hooks.py        # step 4: load it; the API refuses command hooks
└── permission_hooks.py      # step 5: approval gate and a shared audit hook

pip install "promptise>=1.5.0"
python hooks_tour.py
python policy_hooks.py
python shell_hooks.py
python manifest_hooks.py     # uses port 9781
python permission_hooks.py

@process.hooks.on(HookEvent.TASK_COMPLETED)
async def notify(ctx):
    asyncio.create_task(post_to_channel(ctx.data))  # don't hold the worker

"""For comparison: the same password rule as LangChain v1 agent middleware (offline, fake model)."""
from langchain.agents import create_agent
from langchain.agents.middleware import before_model
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage


@before_model(can_jump_to=["end"])
def no_passwords(state, runtime):
    if "password" in state["messages"][-1].content.lower():
        return {"messages": [AIMessage("Password tickets go to the identity team.")],
                "jump_to": "end"}
    return None


model = GenericFakeChatModel(messages=iter([AIMessage("Ticket triaged: P2.")]))
agent = create_agent(model, tools=[], middleware=[no_passwords])

for ticket in ("Please reset my password", "Login page is down"):
    result = agent.invoke({"messages": [{"role": "user", "content": ticket}]})
    print(f"{ticket!r} -> {result['messages'][-1].content}")