Run Outcome

Get a typed terminal outcome from every agent run — no exception juggling

Add return_outcome=True to agent.run() or agent.astart() and the call returns a small frozen RunOutcome telling you exactly how the run ended — instead of raising and making you catch every exception type.

Quick Start

1. Simple usage (bool)

   Branch on outcome.succeeded — no try/except.

   from praisonaiagents import Agent, RunOutcome
   
   agent = Agent(name="Assistant", instructions="Be helpful")
   
   outcome = agent.run("Draft a reply", return_outcome=True)
   
   if outcome.succeeded:
       print(outcome.output)
   else:
       print(f"{outcome.reason}: {outcome.error}")

2. Async with a hard timeout

   astart() accepts a run-level timeout (seconds). When it elapses, reason is hard_timeout.

   import asyncio
   from praisonaiagents import Agent, RunOutcome
   
   agent = Agent(name="Assistant", instructions="Be helpful")
   
   async def main():
       outcome = await agent.astart("Draft a reply", return_outcome=True, timeout=30)
       print(outcome.reason)
   
   asyncio.run(main())

3. Full gateway/bot dispatch

   The headline use case — one branch per terminal reason, no exception chains.

   import asyncio
   from praisonaiagents import Agent, RunOutcome
   
   agent = Agent(name="Assistant", instructions="Be helpful")
   
   async def handle(prompt, run_timeout=30):
       outcome = await agent.astart(prompt, return_outcome=True, timeout=run_timeout)
   
       if outcome.reason == "completed":
           await deliver(outcome.output)
       elif outcome.reason in ("cancelled", "aborted"):
           pass                              # never DLQ, never replay
       elif outcome.reason == "hard_timeout":
           await notify_timeout()
       else:                                 # "failed"
           dlq.enqueue(error=outcome.error)

How It Works

A host receives a request, dispatches it to the agent with return_outcome=True, and picks a channel action from the single reason field.

The eight terminal reasons and the action each one signals:

By default the agent attempts up to 3 auto-continuations before surfacing length_truncated — see Output Continuation. Set ExecutionConfig(output_continuation=False) to surface it on the first truncation.

The last three — content_filtered, refused, length_truncated — come from the LLM’s finish_reason / refusal signal on the response, not from a raised exception. See Provider block outcomes.

Precedence is sticky: a hard_timeout is never silently downgraded by a later cleanup error. A provider block/refusal/truncation outranks failed (so an actionable reason is not masked by a generic empty-result failure) but stays below aborted / cancelled / hard_timeout (host-level lifecycle signals win).

Which reason applies when

A nested inner asyncio.TimeoutError (a tool or handoff that exhausted its own budget) resolves to failed, not hard_timeout. Only the run-level timeout budget produces hard_timeout.

Provider block outcomes

Three reasons come from the LLM’s own response signal instead of an exception — so a run that ends without a usable answer is no longer collapsed into a silent empty completed or a generic failed.

The core classifies the response with classify_finish_reason(finish_reason, refusal): a truthy refusal → refused; finish_reason containing content_filter → content_filtered; length / max_tokens / truncat → length_truncated; a normal stop / tool_calls / unknown → no block. The specific reason surfaces on both RunOutcome.reason and agent.last_stop_reason.

Precedence ladder

A provider block sits above failed but below the host-level lifecycle signals — so an actionable reason is never masked, yet a shutdown/cancel still wins.

A max_steps truncation is sticky and is never downgraded by a later provider block — a provider block only wins when the run did not already hit the step limit.

Configuration Options

RunOutcome is a frozen dataclass. Every field:

Class methods:

from_exception matches by type name: hardtimeout/runtimeout/budgettimeout → hard_timeout; asyncio.CancelledError or supersed/interrupt/cancelled → cancelled; abort/drain/shutdown → aborted; everything else → failed with error=str(exc).

TerminalReason is a Literal["completed", "hard_timeout", "cancelled", "aborted", "failed", "content_filtered", "refused", "length_truncated"] (with a str fallback on Python <3.8), exported from praisonaiagents.agent. The tuple PROVIDER_BLOCK_REASONS = ("content_filtered", "refused", "length_truncated") groups the three provider-block reasons.

New kwargs on run() and astart():

Verified behaviour:

• outcome.reason == "hard_timeout" is authoritative and sticky — never downgraded.
• A nested inner asyncio.TimeoutError resolves to failed, not hard_timeout.
• External asyncio.CancelledError on the enclosing task is re-raised — host shutdown is honoured, not swallowed.
• RunOutcome is frozen — assigning to .reason raises FrozenInstanceError.

Common Patterns

Simple success/failure branch:

outcome = agent.run(prompt, return_outcome=True)
if outcome.succeeded:
    send(outcome.output)
else:
    log(outcome.reason)

Retry ladder — retry only on failed, escalate on hard_timeout, drop the rest:

outcome = await agent.astart(prompt, return_outcome=True, timeout=30)
if outcome.reason == "failed":
    retry(prompt)
elif outcome.reason == "hard_timeout":
    escalate()
elif outcome.reason in ("cancelled", "aborted"):
    pass  # do not retry

Branch on a provider block/refusal/truncation:

outcome = agent.run(prompt, return_outcome=True)
if outcome.reason == "content_filtered":
    notify_user("The provider blocked that request. Try rephrasing.")
elif outcome.reason == "refused":
    notify_user("The model declined. Try a different phrasing or model.")
elif outcome.reason == "length_truncated":
    notify_user("Response was cut off — please increase max_tokens.")

Manual normalisation in a wrapper that owns its own try/except:

from praisonaiagents import RunOutcome

try:
    text = my_custom_call(prompt)
    outcome = RunOutcome.completed(text)
except BaseException as exc:
    outcome = RunOutcome.from_exception(exc)

Best Practices

Migration

The default behaviour of run() / astart() is unchanged — they still return a str and raise on error, so existing code needs no change. Opt in with one flag to collapse three catches into one branch.

# Before — three catches, order matters
try:
    text = await agent.astart(prompt)
    await deliver(text)
except asyncio.TimeoutError:
    await notify_timeout()
except asyncio.CancelledError:
    raise
except Exception as exc:
    dlq.enqueue(error=str(exc))

# After — one branch
outcome = await agent.astart(prompt, return_outcome=True, timeout=run_timeout)
match outcome.reason:
    case "completed":     await deliver(outcome.output)
    case "hard_timeout":  await notify_timeout()
    case "cancelled" | "aborted": pass
    case "failed":        dlq.enqueue(error=outcome.error)

Related