cape.agent.bgtasks: Background tasks for the CAPE agent

This module provides background task support for the CAPE agentic interface (see cape.agent.agentcntl). A background task is a CAPE command, such as cape report -I 100:150, that is run in a child process so that the interactive agent session can continue with other work. The main session checks for completed tasks each time it displays a new prompt.

Two kinds of background task are supported:

  • CLI tasks are full command lines, for example when a user types cape report -I 100:150 & at the agent prompt. The trailing & requests backgrounding, following the usual shell convention.

  • Tool tasks are agent tool calls, e.g. cape_report() with its background option set. These are run by a child copy of this module (python -m cape.agent.bgtasks) so that the full tool result is once again available when the task finishes.

Each task redirects STDOUT and STDERR to a log file in the task folder (<CacheDir>/agent-tasks/ by default, where CacheDir is from the CAPE user configuration), has no STDIN, and is launched in its own session so that Ctrl-C in the main session does not interrupt it. Note that tasks share the file system with the main session; users should avoid running a background task and a foreground command on the same cases at the same time.

class cape.agent.bgtasks.BackgroundTask(task_id: int, cmdlist: list, tool_name: str | None = None)

Record and process handle of one background task

Call:
>>> task = BackgroundTask(task_id, cmdlist, tool_name=None)
Inputs:
task_id: int

Task ID for this session, starting at 1

cmdlist: list[str]

Command being run by the child process

tool_name: {None} | str

Name of agent tool being run; None for CLI tasks

Outputs:
task: BackgroundTask

Record of one background task

cmdlist

list[str] Command being run by the child process

describe() → str

Create a short text description of this task

logfile

str Log file receiving the task’s STDOUT and STDERR

poll() → int | None

Check exit status, None if still running

proc

subprocess.Popen Child process running this task

reaped

bool Whether this task has exited and been processed

resultfile

str | None JSON file of tool results; None for CLI tasks

specfile

str | None JSON file defining a tool task; None for CLI tasks

t0

float Time at which this task was launched

task_id

int Task ID for this session, starting at 1

tool_name

str | None Name of agent tool being run; None for CLI tasks

cape.agent.bgtasks.format_completion_note(task: BackgroundTask, result: dict) → str

Format a terminal note for a completed task

Call:
>>> note = format_completion_note(task, result)
Inputs:
task: BackgroundTask

Record of the completed task

result: dict

Tool-style result of the task

Outputs:
note: str

Note to display to the user

cape.agent.bgtasks.format_history_record(task: BackgroundTask, result: dict) → str

Format a history record for a completed task

Call:
>>> record = format_history_record(task, result)
Inputs:
task: BackgroundTask

Record of the completed task

result: dict

Tool-style result of the task

Outputs:
record: str

Message to append to the conversation history

cape.agent.bgtasks.format_launch_note(task: BackgroundTask) → str

Format a terminal note for a newly launched task

Call:
>>> note = format_launch_note(task)
Inputs:
task: BackgroundTask

Record of the new task

Outputs:
note: str

Note to display to the user

cape.agent.bgtasks.get_task_dir() → str

Get the folder for task logs, creating it if necessary

By default this is agent-tasks/ inside the CAPE CacheDir, but it can be overridden by setting the module-level TASK_DIR.

Call:
>>> taskdir = get_task_dir()
Outputs:
taskdir: str

Name of folder for background task files

cape.agent.bgtasks.launch_cli_task(task_id: int, cmdlist: list) → BackgroundTask

Launch a command line as a background task

Call:
>>> task = launch_cli_task(task_id, cmdlist)
Inputs:
task_id: int

Task ID for this session, starting at 1

cmdlist: list[str]

Command to run in the child process

Outputs:
task: BackgroundTask

Record of the newly launched task

cape.agent.bgtasks.launch_tool_task(task_id: int, tool_name: str, kwargs: dict) → BackgroundTask

Launch an agent tool call as a background task

The tool is run by a child copy of this module (see main()) so that the full tool result is written to a JSON file and recovered by the main session once the task completes.

Call:
>>> task = launch_tool_task(task_id, tool_name, kwargs)
Inputs:
task_id: int

Task ID for this session, starting at 1

tool_name: str

Name of agent tool, e.g. "cape_report"

kwargs: dict

Keyword arguments to the tool

Outputs:
task: BackgroundTask

Record of the newly launched task

cape.agent.bgtasks.main(argv: list | None = None) → int

Run one agent tool from a JSON spec file

This is the entry point used by child processes created by launch_tool_task(); it runs the tool and writes its result to the JSON file named in the spec.

Call:
>>> ierr = main(argv=None)
Inputs:
argv: {None} | list[str]

Command-line args, the last of which is the spec file; defaults to sys.argv[1:]

Outputs:
ierr: int

Return code of the tool call

cape.agent.bgtasks.poll_finished(tasks: list) → list

Check for newly completed background tasks

Call:
>>> finished = poll_finished(tasks)
Inputs:
tasks: list[BackgroundTask]

List of tasks launched during this session

Outputs:
finished: list[tuple]

Newly completed (task, result) pairs; result is a dict in the style of a normal tool result

cape.agent.bgtasks.read_task_result(task: BackgroundTask, returncode: int) → dict

Assemble a tool-style result dict for a completed task

For tool tasks the result written by the child process is read back; for other tasks (or a tool task whose child died early) a result is synthesized from the exit code and a tail of the log file.

Call:
>>> result = read_task_result(task, returncode)
Inputs:
task: BackgroundTask

Record of the completed task

returncode: int

Exit status of the child process

Outputs:
result: dict

Tool-style result, including success and returncode