JOB_COMPLETED Hook

React to background job completion — success or failure — with a lightweight event hook

HookEvent.JOB_COMPLETED fires from BackgroundJobManager.start_job whenever any managed background job reaches a terminal state — completed or failed. Register a hook to log completions, send alerts, or trigger follow-up work.

from praisonaiagents import Agent

agent = Agent(name="Coordinator", instructions="Delegate long tasks to background workers")
agent.start("Analyse this repo and email me when done")

The user starts background work; your JOB_COMPLETED hook runs when the subagent finishes or fails.

How It Works

Quick Start

1. Create a hook registry

   from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult
   from praisonaiagents.hooks import JobCompletedInput
   
   registry = HookRegistry()

2. Register a JOB_COMPLETED handler

   @registry.on(HookEvent.JOB_COMPLETED)
   def log_completion(data: JobCompletedInput) -> HookResult:
       with open("job_completions.log", "a") as f:
           f.write(
               f"{data.job_id}\t{data.status}\t{data.deliver or '-'}\t{data.error or ''}\n"
           )
       return HookResult(decision="allow")

3. Attach to your agent

   from praisonaiagents import Agent
   
   agent = Agent(
       name="SupportBot",
       instructions="Help users.",
       hooks=registry,
   )

How It Works

The user delegates work to a background job; when the job ends, the hook fires with the outcome.

JobCompletedInput Fields

JobCompletedInput carries everything about the finished job, including the delivery context if one was set. All origin fields are strings — empty ("") when not set.

Origin fields (session_id, deliver, platform, chat_id, thread_id) are always strings — an empty string when not set.

Behaviour Notes

• Fires for every managed background job, not only those launched via spawn_subagent(background=True) — BackgroundJobManager.start_job emits it directly at the terminal path.
• Fires on both success and failure. Check data.status ("completed" or "failed") to distinguish.
• Does not fire for CANCELLED or LOST. The enum documents “finished ok or error”; that contract was deliberately not widened. This holds even when a running job is cancelled — its late outcome is discarded.
• Best-effort. A raising hook never crashes the background worker — exceptions are caught and logged.
• Sits alongside SCHEDULE_TRIGGER. Both fire in the background-job lifecycle and follow the same hook shape.

@registry.on(HookEvent.JOB_COMPLETED)
def alert_on_failure(data: JobCompletedInput) -> HookResult:
    if data.status == "failed":
        send_alert(f"Job {data.job_id} failed: {data.error}")
    return HookResult(decision="allow")

Logging Example

The shortest copy-paste hook that logs every completion to a file:

import datetime
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult
from praisonaiagents.hooks import JobCompletedInput

registry = HookRegistry()

@registry.on(HookEvent.JOB_COMPLETED)
def log_job(data: JobCompletedInput) -> HookResult:
    ts = datetime.datetime.utcnow().isoformat()
    line = f"[{ts}] job={data.job_id} status={data.status} deliver={data.deliver!r}\n"
    if data.error:
        line += f"  error: {data.error}\n"
    with open("jobs.log", "a") as f:
        f.write(line)
    return HookResult(decision="allow")

Best Practices

import concurrent.futures

_pool = concurrent.futures.ThreadPoolExecutor(max_workers=4)

@registry.on(HookEvent.JOB_COMPLETED)
def notify_async(data: JobCompletedInput) -> HookResult:
    _pool.submit(send_slack_message, data.job_id, data.status)
    return HookResult(decision="allow")

@registry.on(HookEvent.JOB_COMPLETED)
def safe_hook(data: JobCompletedInput) -> HookResult:
    try:
        record_to_db(data)
    except Exception as exc:
        import logging
        logging.getLogger(__name__).error("hook failed: %s", exc)
    return HookResult(decision="allow")

Related