ctx.stream

class StreamEmitter:
    def emit(self, event_type: str, data: dict | str) -> None: ...
    def progress(self, content: str, *, tool_name: str | None = None) -> None: ...
    def intent(self, content: str) -> None: ...

ctx.stream.emit("custom-phase", {"step": 3, "total": 10})
ctx.stream.emit("debug", "Processing complete")

ctx.stream.progress("Starting analysis...")
ctx.stream.progress("Fetching repository data", tool_name="GitHub")
ctx.stream.progress("Analysing code patterns", tool_name="Analyser")

ctx.stream.intent("Discovering repository structure")
ctx.stream.intent("Identifying security issues")
ctx.stream.intent("Generating recommendations")

def execute(prompt, ctx):
    ctx.stream.progress("Phase 1: Parsing input")
    config = parse_input(prompt)

    ctx.stream.progress("Phase 2: Fetching data", tool_name="GitHub")
    data = ctx.tools.call("fetch_repo", config)

    ctx.stream.progress("Phase 3: Analysing", tool_name="LLM")
    analysis = ctx.llm.generate(...)

    ctx.stream.progress("Phase 4: Finalising")
    return ok({"result": analysis.text})

def execute(prompt, ctx):
    ctx.stream.intent("Understanding task requirements")
    requirements = extract_requirements(prompt)

    ctx.stream.intent("Planning approach")
    plan = create_plan(requirements)

    ctx.stream.intent("Executing plan")
    for step in plan.steps:
        ctx.stream.progress(f"Step {step.number}: {step.description}")
        execute_step(step)

    ctx.stream.intent("Finalising results")
    return ok({"completed": True})

def execute(prompt, ctx):
    ctx.stream.progress("Initialising", tool_name="Setup")

    ctx.stream.progress("Querying database", tool_name="PostgreSQL")
    rows = ctx.tools.call("query", {"sql": "SELECT ..."})

    ctx.stream.progress("Processing results", tool_name="Processor")
    processed = [transform(r) for r in rows]

    ctx.stream.progress("Storing analysis", tool_name="Storage")
    ctx.http.fetch(..., method="POST", body=json.dumps(processed))

    ctx.stream.progress("Complete", tool_name="Setup")
    return ok({"count": len(processed)})

def execute(prompt, ctx):
    # Safe wrapper
    def progress(msg, tool=None):
        if ctx.stream:
            ctx.stream.progress(msg, tool_name=tool)

    progress("Starting...")

    # Work...

    progress("Complete")
    return ok({"done": True})

ctx.stream.progress("Starting LLM call...")  # Sent immediately

# LLM call blocks until response; progress already sent
result = ctx.llm.generate(messages, model="claude-sonnet-4-6")

# Back in Python
ctx.stream.progress("LLM complete")  # Sent now