mirror of
https://github.com/MervinPraison/PraisonAI.git
synced 2026-11-07 17:33:33 -10:00
feat(ts): build the Task engine behind 28 accepted-and-ignored options (#4838)
* feat(ts): build the Task engine behind 28 accepted-and-ignored options
Task accepted 32 options and acted on none of them. This lands the behaviour
for 28, as engine modules under src/agent/engine/ with types.ts delegating to
them: retries and backoff, dependency-failure policy, context assembly,
routing and branching, loop expansion, input-file fan-out, multimodal message
building, handler execution, output shaping, per-task memory, knowledge, hooks
and caching.
The ledger is deliberately unchanged, and that is the point worth reading.
Every one of these options only takes effect when the team runner calls the
code, and it does not: grepping for the new methods outside the engine and
types.ts finds zero call sites. So passing `handler` or `loopOver` to a Task
today still does nothing, exactly as before. Shrinking the ledger now would
drop the count by 28 while nothing changed for anyone using the SDK -- the
precise move the ratchet cannot detect, arrived at honestly. The agent that
wrote this flagged the same thing in its own report.
They are closed when team.ts calls them, in the change that wires the runner;
until then this is groundwork with tests, not a closed gap.
What the runner needs, all re-exported from engine/index.ts: startTaskOf and
linkPreviousTasks before the run, planTaskRun to drive the loop, and per task
needsRun, notifyStart, resolveAgent, runHandler, buildContext, memoryContext,
knowledgeContext, buildMessageContent, the cache lookup, buildOutput in place
of the inline literal at team.ts:307, storeInMemory after, waitBeforeRetry and
continuesOnDependencyFailure on failure, and routeAfter with expandLoop and
expandFromInputFile at plan time.
Four options are not implemented and stay open on their own merits: autonomy,
web, reflection and planning are stored on the Python Task and never read
anywhere outside task.py's assignment, so implementing them would be inventing
behaviour the reference does not have.
One sub-gap inside a closed option, documented in the module header: Python
decodes a local .mp4 with OpenCV and appends a frame per second. There is no
decoder here, so a local video contributes a text note naming the file rather
than an unusable data URI. Images and URLs are fully ported.
Also fixed: the ledger assertion in parity-task.test.ts derived its expectation
from a hard-coded array, so it would have drifted the moment the ledger changed.
It now reads unhonouredFor('Task.__init__') instead.
42 new tests. tsc clean, 2,118 passing.
Co-Authored-By: Claude Opus 5 <[email protected]>
* fix(ts): close reviewer-flagged gaps in the Task engine
- CSV: parse whole-document records so quoted newlines and doubled quotes
survive (RFC 4180 + backslash escape), instead of splitting physical lines
- loop/inputFile fan-out: children inherit the parent execution contract
(handler, output, retries, hooks, memory, knowledge, caching, tools, images)
- output: only assign outputJson/outputPydantic when the parsed value is an
object; record a non-fatal error for arrays/scalars/null
- knowledge: accept the canonical Knowledge `{ results: [...] }` shape as well
as the KnowledgeBase array shape
- messages: map image extensions to registered MIME types (.jpg → image/jpeg)
- agent: record construction failure on nonFatalErrors before falling back
- callback: always pass metadata; JS callbacks ignore the extra argument
- fix pre-existing timing flake in workflow duration test (49ms vs >=50ms)
- +6 regression tests
🤖 Generated with [Claude Code](https://claude.ai/code)
Co-authored-by: Mervin Praison <[email protected]>
---------
Co-authored-by: Claude Opus 5 <[email protected]>
Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: Mervin Praison <[email protected]>
This commit is contained in:
18 files changed
+2619
-14
No files matched your search
@@ -0,0 +1,73 @@
|
||||
/**
|
||||
* The Task execution engine: the behaviour behind the `Task` options that only
|
||||
* mean something to the loop that runs tasks.
|
||||
*
|
||||
* `Task` exposes most of this as methods (`buildContext`, `nextTaskFor`,
|
||||
* `expandLoop`, `runHandler`, ...); the functions here are the ones a runner
|
||||
* needs across a whole task *list* rather than for one task.
|
||||
*/
|
||||
|
||||
export { evaluateWhen } from './conditions';
|
||||
|
||||
export {
|
||||
MAX_RETRY_DELAY_SECONDS,
|
||||
continuesOnDependencyFailure,
|
||||
needsRun,
|
||||
retryDelaySeconds,
|
||||
sleepSeconds,
|
||||
} from './task-retry';
|
||||
export type { RetryPolicyLike } from './task-retry';
|
||||
|
||||
export { dependsOnPending, planTaskBatches, planTaskRun } from './task-schedule';
|
||||
export type { SchedulableTask, TaskBatch } from './task-schedule';
|
||||
|
||||
export { CONTEXT_HEADER, buildTaskContext, contextOutputs, renderValidationFeedback } from './task-context';
|
||||
export type { PreviousOutput } from './task-context';
|
||||
|
||||
export {
|
||||
EXIT,
|
||||
decisionRoute,
|
||||
evaluateTaskWhen,
|
||||
hasWhenRouting,
|
||||
linkPreviousTasks,
|
||||
resolveNextTask,
|
||||
startTaskOf,
|
||||
} from './task-routing';
|
||||
export type { RouteDecision, RoutingContext } from './task-routing';
|
||||
|
||||
export { inputFileRows, inputFileTaskConfigs, parseCsvLine } from './task-input-file';
|
||||
export type { FileReader } from './task-input-file';
|
||||
|
||||
export { LOOP_INDEX_VAR, isLoopTask, loopTaskConfigs } from './task-loop';
|
||||
|
||||
export { buildMultimodalContent, videoNote } from './task-messages';
|
||||
export type { ImageFileSystem, MessageContentPart } from './task-messages';
|
||||
|
||||
export { buildTaskOutput, cleanJsonOutput, resolveOutputConfig } from './task-output';
|
||||
export type { ResolvedOutputConfig } from './task-output';
|
||||
|
||||
export { runTaskHandler } from './task-handler';
|
||||
export type { HandlerContext, HandlerResult, HandlerStepResult } from './task-handler';
|
||||
|
||||
export { agentOptionsFor, resolveTaskAgent } from './task-agent';
|
||||
export type { AgentFactory, ResolveAgentOptions } from './task-agent';
|
||||
|
||||
export { buildMemoryContext, initializeTaskMemory, memoryConfigOf, storeTaskOutput } from './task-memory';
|
||||
export type { MemoryFactory, TaskMemoryStore } from './task-memory';
|
||||
|
||||
export { buildKnowledgeContext, cacheKey, resolveTaskCache, resolveTaskHooks } from './task-features';
|
||||
export type { KnowledgeSearcher } from './task-features';
|
||||
|
||||
export { ResponseCache } from './types';
|
||||
export type {
|
||||
EngineTask,
|
||||
ExecutionEnv,
|
||||
GuardrailJudgeFactory,
|
||||
GuardrailJudgeResult,
|
||||
ResolvedExecution,
|
||||
TaskHandlerContext,
|
||||
TaskHooks,
|
||||
TaskInput,
|
||||
TaskRunResult,
|
||||
TeamHooks,
|
||||
} from './types';
|
||||
@@ -0,0 +1,68 @@
|
||||
/**
|
||||
* `agentConfig`: building the executing agent for a task that has none.
|
||||
*
|
||||
* Port of `praisonaiagents/agents/agents.py::_create_agent_from_config` and
|
||||
* `workflows.py::_create_step_agent`. The task's own `agent` always wins; with
|
||||
* no agent but an `agentConfig`, an agent is constructed from it (defaulting
|
||||
* `name` to `<task name>Agent` and `llm` to the run's default model, and taking
|
||||
* the task's `tools` when it has any); with neither, the caller's default agent
|
||||
* is used. Python assigns the constructed agent back onto the task, and so does
|
||||
* this.
|
||||
*/
|
||||
|
||||
import type { Task } from '../types';
|
||||
|
||||
/** Builds an agent from constructor options; injectable so tests need no LLM. */
|
||||
export type AgentFactory = (options: Record<string, unknown>) => unknown;
|
||||
|
||||
export interface ResolveAgentOptions {
|
||||
/** Used when the task names no agent and carries no `agentConfig`. */
|
||||
defaultAgent?: unknown;
|
||||
/** Applied as `llm` when `agentConfig` does not name one. */
|
||||
defaultLlm?: string;
|
||||
/** Shared memory handed to a constructed agent. */
|
||||
memory?: unknown;
|
||||
/** Overrides the real `Agent` constructor (tests, or an alternate class). */
|
||||
createAgent?: AgentFactory;
|
||||
}
|
||||
|
||||
/** The `Agent` constructor, required lazily so this module has no import cycle. */
|
||||
function defaultAgentFactory(options: Record<string, unknown>): unknown {
|
||||
// eslint-disable-next-line @typescript-eslint/no-var-requires
|
||||
const { Agent } = require('../simple') as { Agent: new (o: Record<string, unknown>) => unknown };
|
||||
return new Agent(options);
|
||||
}
|
||||
|
||||
/** The constructor options `agentConfig` resolves to, before the agent is built. */
|
||||
export function agentOptionsFor(task: Task, options: ResolveAgentOptions = {}): Record<string, unknown> | undefined {
|
||||
if (!task.agentConfig) return undefined;
|
||||
const built: Record<string, unknown> = { ...task.agentConfig };
|
||||
delete built.verbose;
|
||||
if (built.name === undefined) built.name = task.name ? `${task.name}Agent` : 'TaskAgent';
|
||||
if (built.llm === undefined && options.defaultLlm !== undefined) built.llm = options.defaultLlm;
|
||||
if (task.tools.length > 0) built.tools = [...task.tools];
|
||||
if (options.memory !== undefined && built.memory === undefined) built.memory = options.memory;
|
||||
return built;
|
||||
}
|
||||
|
||||
/**
|
||||
* The agent that should execute `task`.
|
||||
*
|
||||
* A constructed agent is assigned to `task.agent`, matching Python. Construction
|
||||
* failures fall back to `defaultAgent` rather than aborting the run.
|
||||
*/
|
||||
export function resolveTaskAgent(task: Task, options: ResolveAgentOptions = {}): unknown | undefined {
|
||||
if (task.agent) return task.agent;
|
||||
const built = agentOptionsFor(task, options);
|
||||
if (!built) return options.defaultAgent;
|
||||
const factory = options.createAgent ?? defaultAgentFactory;
|
||||
try {
|
||||
const agent = factory(built);
|
||||
task.agent = agent;
|
||||
return agent;
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
task.nonFatalErrors.push(`resolve_agent: ${message}`);
|
||||
return options.defaultAgent;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
/**
|
||||
* Inter-task context assembly (Python `Process._build_task_context`).
|
||||
*
|
||||
* `retainFullContext` is the switch: with it every upstream result is folded
|
||||
* into the prompt; without it only the most recent one is, which is the whole
|
||||
* point of the flag (Python's comment calls it "Original behavior" vs "New
|
||||
* behavior"). Validation feedback from a failed guardrail is prepended so a
|
||||
* retry sees why the last attempt was rejected.
|
||||
*/
|
||||
|
||||
import type { Task, TaskOutput } from '../types';
|
||||
|
||||
/** One upstream result, in run order. */
|
||||
export interface PreviousOutput {
|
||||
name: string;
|
||||
raw: string;
|
||||
}
|
||||
|
||||
export const CONTEXT_HEADER = '\nInput data from previous tasks:';
|
||||
|
||||
function rawOf(result: TaskOutput | string | null | undefined): string | undefined {
|
||||
if (result === null || result === undefined) return undefined;
|
||||
if (typeof result === 'string') return result.length > 0 ? result : undefined;
|
||||
return result.raw && result.raw.length > 0 ? result.raw : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* The results the task's own `context` list contributes, in declaration order.
|
||||
* String context items are literal input; Task items contribute their result.
|
||||
*/
|
||||
export function contextOutputs(task: Task): PreviousOutput[] {
|
||||
const out: PreviousOutput[] = [];
|
||||
for (const item of task.context) {
|
||||
if (typeof item === 'string') {
|
||||
if (item.trim().length > 0) out.push({ name: 'input', raw: item });
|
||||
continue;
|
||||
}
|
||||
const raw = rawOf(item.result);
|
||||
const name = item.name ?? item.description;
|
||||
if (raw !== undefined && name !== task.name) out.push({ name: name || 'context', raw });
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Render `validationFeedback` the way Python's retry prompt does. */
|
||||
export function renderValidationFeedback(feedback: unknown): string | undefined {
|
||||
if (feedback === undefined || feedback === null) return undefined;
|
||||
if (typeof feedback === 'string') {
|
||||
return `\nPrevious attempt failed validation with reason: ${feedback}`
|
||||
+ '\nPlease try again with a different approach based on this feedback.\n';
|
||||
}
|
||||
if (typeof feedback !== 'object') return undefined;
|
||||
const f = feedback as Record<string, unknown>;
|
||||
const reason = f.validation_response ?? f.validationResponse ?? f.reason;
|
||||
let text = `\nPrevious attempt failed validation with reason: ${String(reason ?? '')}`;
|
||||
const validated = f.validated_task ?? f.validatedTask;
|
||||
if (validated) text += `\nValidated task: ${String(validated)}`;
|
||||
const details = f.validation_details ?? f.validationDetails;
|
||||
if (details) text += `\nValidation feedback: ${String(details)}`;
|
||||
const rejected = f.rejected_output ?? f.rejectedOutput;
|
||||
if (rejected) text += `\nRejected output: ${String(rejected)}`;
|
||||
text += '\nPlease try again with a different approach based on this feedback.\n';
|
||||
return text;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the context block for `task`.
|
||||
*
|
||||
* @param task - the task about to run.
|
||||
* @param previous - upstream results in run order (workflow `nextTasks` edges).
|
||||
* The runner supplies these; the task's own `context` list is read from the
|
||||
* task itself.
|
||||
*
|
||||
* Consuming `validationFeedback` clears it, matching Python, so the feedback is
|
||||
* injected exactly once.
|
||||
*/
|
||||
export function buildTaskContext(task: Task, previous: readonly PreviousOutput[] = []): string {
|
||||
const feedback = renderValidationFeedback(task.validationFeedback);
|
||||
const ctx = contextOutputs(task);
|
||||
if (feedback !== undefined) task.validationFeedback = undefined;
|
||||
|
||||
if (previous.length === 0 && ctx.length === 0) return feedback ?? '';
|
||||
|
||||
let text = feedback === undefined ? CONTEXT_HEADER : feedback + CONTEXT_HEADER;
|
||||
|
||||
if (task.retainFullContext) {
|
||||
for (const p of previous) text += `\n${p.name}: ${p.raw}`;
|
||||
for (const c of ctx) text += `\n${c.name}: ${c.raw}`;
|
||||
return text;
|
||||
}
|
||||
|
||||
// Only the most recent upstream result, and the most recent context result.
|
||||
const lastPrevious = previous[previous.length - 1];
|
||||
if (lastPrevious) text += `\n${lastPrevious.name}: ${lastPrevious.raw}`;
|
||||
const lastContext = ctx[ctx.length - 1];
|
||||
if (lastContext) text += `\n${lastContext.name}: ${lastContext.raw}`;
|
||||
return text;
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
/**
|
||||
* Per-task feature configs: `hooks`, `caching` and `knowledge`.
|
||||
*
|
||||
* These are the task-scoped counterparts of the team-level options of the same
|
||||
* names; `caching` resolves through the package's canonical `resolve_caching`
|
||||
* (Python `config/param_resolver.py`).
|
||||
*
|
||||
* - `hooks` — an `onTaskStart` / `onTaskComplete` pair fired around this task
|
||||
* only (Python `WorkflowHooksConfig.on_step_start` / `on_step_complete`);
|
||||
* snake_case and the workflow's `on_step_*` spelling are both accepted.
|
||||
* - `caching` — a per-task response cache keyed by agent + prompt, so a repeat
|
||||
* of the same prompt is answered from the cache instead of the model.
|
||||
* - `knowledge` — literal text, a list of documents, or any object exposing
|
||||
* `search(query, limit)` (a `KnowledgeBase`); the matches are rendered into
|
||||
* the task prompt.
|
||||
*/
|
||||
|
||||
import { resolve_caching } from '../../config';
|
||||
import type { CachingConfig } from '../../config';
|
||||
import type { Task, TaskOutput } from '../types';
|
||||
import { ResponseCache } from './types';
|
||||
import type { TaskHooks } from './types';
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function asFunction(value: unknown): ((...args: never[]) => unknown) | undefined {
|
||||
return typeof value === 'function' ? value as (...args: never[]) => unknown : undefined;
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------- hooks
|
||||
|
||||
/**
|
||||
* Resolve a task's `hooks` option. Returns `undefined` when it carries no
|
||||
* callable, so a task without hooks costs nothing at run time.
|
||||
*/
|
||||
export function resolveTaskHooks(input: unknown): TaskHooks | undefined {
|
||||
if (input === undefined || input === null || input === false) return undefined;
|
||||
if (!isRecord(input)) return undefined;
|
||||
|
||||
const start = asFunction(input.onTaskStart ?? input.on_task_start ?? input.onStepStart ?? input.on_step_start);
|
||||
const complete = asFunction(input.onTaskComplete ?? input.on_task_complete ?? input.onStepComplete ?? input.on_step_complete);
|
||||
if (!start && !complete) return undefined;
|
||||
|
||||
const hooks: TaskHooks = {};
|
||||
if (start) hooks.onTaskStart = start as (task: Task, index: number) => unknown;
|
||||
if (complete) hooks.onTaskComplete = complete as (task: Task, output: TaskOutput) => unknown;
|
||||
return hooks;
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------ caching
|
||||
|
||||
/** Resolve a task's `caching` option to a cache, or `undefined` when disabled. */
|
||||
export function resolveTaskCache(input: unknown): ResponseCache | undefined {
|
||||
if (input === undefined || input === null || input === false) return undefined;
|
||||
const config = resolve_caching(input as boolean | string | CachingConfig | undefined);
|
||||
if (!config || config.enabled === false) return undefined;
|
||||
return new ResponseCache(config.ttl ?? 0);
|
||||
}
|
||||
|
||||
/** The cache key a task's prompt uses: the agent name plus the prompt text. */
|
||||
export function cacheKey(agentName: string, prompt: string): string {
|
||||
return `${agentName}::${prompt}`;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------- knowledge
|
||||
|
||||
/** Anything that can answer a knowledge query (`KnowledgeBase` or `Knowledge`). */
|
||||
export interface KnowledgeSearcher {
|
||||
search(query: string, ...args: unknown[]): unknown;
|
||||
}
|
||||
|
||||
function isSearcher(value: unknown): value is KnowledgeSearcher {
|
||||
return typeof value === 'object' && value !== null && typeof (value as KnowledgeSearcher).search === 'function';
|
||||
}
|
||||
|
||||
/**
|
||||
* Pull the text out of a search hit regardless of its shape. `KnowledgeBase`
|
||||
* returns `{ document: { content } }` items; the canonical `Knowledge` returns
|
||||
* `SearchResultItem` objects carrying `text`. Both spellings, plus a plain
|
||||
* string, are read here.
|
||||
*/
|
||||
function hitText(hit: unknown): string {
|
||||
if (typeof hit === 'string') return hit;
|
||||
if (!isRecord(hit)) return '';
|
||||
const doc = hit.document;
|
||||
if (isRecord(doc) && typeof doc.content === 'string') return doc.content;
|
||||
if (typeof hit.text === 'string') return hit.text;
|
||||
if (typeof hit.content === 'string') return hit.content;
|
||||
return '';
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalise a searcher's return value to a list of hits. `KnowledgeBase`
|
||||
* returns the array directly; `Knowledge` returns `{ results: [...] }`.
|
||||
*/
|
||||
function hitsOf(result: unknown): unknown[] {
|
||||
if (Array.isArray(result)) return result;
|
||||
if (isRecord(result) && Array.isArray((result as { results?: unknown }).results)) {
|
||||
return (result as { results: unknown[] }).results;
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
/**
|
||||
* Render the task's `knowledge` into a prompt block.
|
||||
*
|
||||
* A string or list of strings is literal knowledge and is included verbatim; an
|
||||
* object with `search` is queried with `query`. Returns `''` when the task has
|
||||
* no knowledge or nothing matched.
|
||||
*/
|
||||
export async function buildKnowledgeContext(
|
||||
task: Task,
|
||||
query: string,
|
||||
limit = 5
|
||||
): Promise<string> {
|
||||
const knowledge = task.knowledge;
|
||||
if (knowledge === undefined || knowledge === null) return '';
|
||||
|
||||
if (typeof knowledge === 'string') return knowledge;
|
||||
if (Array.isArray(knowledge)) {
|
||||
const texts = knowledge.filter((k): k is string => typeof k === 'string');
|
||||
return texts.join('\n');
|
||||
}
|
||||
if (isSearcher(knowledge)) {
|
||||
try {
|
||||
const hits = hitsOf(await knowledge.search(query, limit));
|
||||
if (hits.length === 0) return '';
|
||||
return hits.map(hitText).filter((t) => t.length > 0).join('\n');
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
task.nonFatalErrors.push(`knowledge search: ${message}`);
|
||||
return '';
|
||||
}
|
||||
}
|
||||
return '';
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
/**
|
||||
* `handler`: running a plain function in place of an agent.
|
||||
*
|
||||
* Port of the handler branch in
|
||||
* `praisonaiagents/workflows/workflows.py::_run_step`. The handler is called
|
||||
* with a workflow context; a `StepResult`-shaped return contributes its
|
||||
* `output`, merges its `variables` into the run scope and may stop the
|
||||
* workflow, while any other return value is stringified. A throwing handler
|
||||
* fails the step instead of the whole run.
|
||||
*/
|
||||
|
||||
import type { Task } from '../types';
|
||||
|
||||
/** Python `WorkflowContext`: what a handler is handed. */
|
||||
export interface HandlerContext {
|
||||
/** The run's original input. */
|
||||
input?: string;
|
||||
/** Raw output of the task that ran before this one. */
|
||||
previousResult?: string;
|
||||
/** Name of the task being run. */
|
||||
currentStep?: string;
|
||||
/** Run variables (a copy, as in Python). */
|
||||
variables: Record<string, unknown>;
|
||||
/** The task itself, for handlers that want it. */
|
||||
task?: Task;
|
||||
}
|
||||
|
||||
/** Python `StepResult`. */
|
||||
export interface HandlerStepResult {
|
||||
output?: unknown;
|
||||
variables?: Record<string, unknown>;
|
||||
stopWorkflow?: boolean;
|
||||
/** snake_case accepted, as Python's dataclass field is `stop_workflow`. */
|
||||
stop_workflow?: boolean;
|
||||
}
|
||||
|
||||
/** What running a handler produced. */
|
||||
export interface HandlerResult {
|
||||
success: boolean;
|
||||
output: string | null;
|
||||
/** Variables the handler asked to publish into the run scope. */
|
||||
variables?: Record<string, unknown>;
|
||||
/** Python `StepResult.stop_workflow`: end the run after this step. */
|
||||
stop: boolean;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
function isStepResult(value: unknown): value is HandlerStepResult {
|
||||
if (typeof value !== 'object' || value === null || Array.isArray(value)) return false;
|
||||
const v = value as Record<string, unknown>;
|
||||
return 'output' in v || 'variables' in v || 'stopWorkflow' in v || 'stop_workflow' in v;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run `task.handler` with `context`.
|
||||
*
|
||||
* @returns `undefined` when the task has no handler, so the caller falls
|
||||
* through to the agent path.
|
||||
*/
|
||||
export async function runTaskHandler(
|
||||
task: Task,
|
||||
context: HandlerContext
|
||||
): Promise<HandlerResult | undefined> {
|
||||
if (!task.handler) return undefined;
|
||||
try {
|
||||
const returned = await Promise.resolve(task.handler(context));
|
||||
if (isStepResult(returned)) {
|
||||
const step = returned;
|
||||
const result: HandlerResult = {
|
||||
success: true,
|
||||
output: step.output === undefined || step.output === null ? null : String(step.output),
|
||||
stop: Boolean(step.stopWorkflow ?? step.stop_workflow),
|
||||
};
|
||||
if (step.variables) result.variables = { ...step.variables };
|
||||
return result;
|
||||
}
|
||||
return {
|
||||
success: true,
|
||||
output: returned === undefined || returned === null ? null : String(returned),
|
||||
stop: false,
|
||||
};
|
||||
} catch (err) {
|
||||
return {
|
||||
success: false,
|
||||
output: null,
|
||||
stop: false,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
/**
|
||||
* `inputFile` fan-out (Python `Process._create_loop_subtasks`).
|
||||
*
|
||||
* A task with an `inputFile` is a template: each CSV row (or each non-empty
|
||||
* line of a text file) becomes one subtask whose description is the parent
|
||||
* description followed by the row. The subtasks are chained with `nextTasks`,
|
||||
* the first is flagged `isStart`, and the last inherits the parent's
|
||||
* `nextTasks` so the workflow rejoins its original successor.
|
||||
*
|
||||
* The module returns `TaskConfig` values rather than `Task` instances so it
|
||||
* never has to import the `Task` class at run time.
|
||||
*/
|
||||
|
||||
import * as fs from 'fs';
|
||||
import * as path from 'path';
|
||||
import type { Task, TaskConfig } from '../types';
|
||||
import { inheritedExecutionConfig } from './task-loop';
|
||||
|
||||
/**
|
||||
* Parse a whole CSV document into logical records. `"` quotes a field, a
|
||||
* doubled `""` inside a quoted field is a literal quote (RFC 4180), a `\` escapes
|
||||
* the next character outside quotes (Python's `escapechar='\\'`), and a newline
|
||||
* inside quotes stays part of the field rather than starting a new record.
|
||||
* Returns one string[] per logical record.
|
||||
*/
|
||||
export function parseCsv(contents: string): string[][] {
|
||||
const records: string[][] = [];
|
||||
let record: string[] = [];
|
||||
let field = '';
|
||||
let quoted = false;
|
||||
let started = false;
|
||||
const pushField = () => { record.push(field); field = ''; };
|
||||
const pushRecord = () => { pushField(); records.push(record); record = []; started = false; };
|
||||
|
||||
for (let i = 0; i < contents.length; i++) {
|
||||
const ch = contents[i];
|
||||
if (quoted) {
|
||||
if (ch === '"') {
|
||||
if (contents[i + 1] === '"') { field += '"'; i++; }
|
||||
else quoted = false;
|
||||
} else {
|
||||
field += ch;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (ch === '\\' && i + 1 < contents.length) { field += contents[++i]; started = true; continue; }
|
||||
if (ch === '"') { quoted = true; started = true; continue; }
|
||||
if (ch === ',') { pushField(); started = true; continue; }
|
||||
if (ch === '\r') { continue; }
|
||||
if (ch === '\n') { if (started || field.length > 0) pushRecord(); continue; }
|
||||
field += ch;
|
||||
started = true;
|
||||
}
|
||||
if (started || field.length > 0) pushRecord();
|
||||
return records;
|
||||
}
|
||||
|
||||
/**
|
||||
* Split one CSV line, honouring `"` quoting and `""` doubled-quote escaping.
|
||||
* Kept for callers that already hold a single physical record; prefer
|
||||
* `parseCsv` for whole files so quoted newlines survive.
|
||||
*/
|
||||
export function parseCsvLine(line: string): string[] {
|
||||
const [record] = parseCsv(line);
|
||||
return record ?? [''];
|
||||
}
|
||||
|
||||
/**
|
||||
* The row texts an input file contributes. A CSV row of two or more fields
|
||||
* becomes Python's `"Question: <a>\nAnswer: <rest>"` pair; a single field, or a
|
||||
* line of a text file, is used verbatim. Empty rows are dropped.
|
||||
*/
|
||||
export function inputFileRows(contents: string, extension: string): string[] {
|
||||
if (extension.toLowerCase() !== '.csv') {
|
||||
return contents.split(/\r?\n/).map((l) => l.trim()).filter((l) => l.length > 0);
|
||||
}
|
||||
const rows: string[] = [];
|
||||
for (const fields of parseCsv(contents)) {
|
||||
if (fields.length === 0) continue;
|
||||
let text = (fields[0] ?? '').trim();
|
||||
if (fields.length > 1) {
|
||||
const question = (fields[0] ?? '').trim();
|
||||
const answer = fields.slice(1).map((f) => f.trim()).join(',');
|
||||
text = `Question: ${question}\nAnswer: ${answer}`;
|
||||
}
|
||||
if (text.length > 0) rows.push(text);
|
||||
}
|
||||
return rows;
|
||||
}
|
||||
|
||||
/** How the parent task's file is read; injectable so tests need no disk. */
|
||||
export type FileReader = (filePath: string) => string;
|
||||
|
||||
const readFromDisk: FileReader = (filePath) => fs.readFileSync(filePath, 'utf8');
|
||||
|
||||
/**
|
||||
* Build the subtask configs for `task`'s `inputFile`.
|
||||
*
|
||||
* @param decisionMode - Python's `decision_mode`: subtasks become `decision`
|
||||
* tasks carrying a `done`/`retry`/`exit` condition table, as the workflow
|
||||
* engine does for a start task.
|
||||
* @returns `[]` when the task has no `inputFile`.
|
||||
* @throws when the file cannot be read — the caller marks the task failed, as
|
||||
* Python does.
|
||||
*/
|
||||
export function inputFileTaskConfigs(
|
||||
task: Task,
|
||||
options: { decisionMode?: boolean; readFile?: FileReader } = {}
|
||||
): TaskConfig[] {
|
||||
if (!task.inputFile) return [];
|
||||
const readFile = options.readFile ?? readFromDisk;
|
||||
const rows = inputFileRows(readFile(task.inputFile), path.extname(task.inputFile));
|
||||
if (rows.length === 0) return [];
|
||||
|
||||
const inheritedNext = task.nextTasks.length > 0 ? [...task.nextTasks] : [];
|
||||
const inheritedExecution = inheritedExecutionConfig(task);
|
||||
const nameFor = (index: number, text: string): string =>
|
||||
task.name ? `${task.name}_${index + 1}` : text;
|
||||
|
||||
const names = rows.map((text, i) => nameFor(i, text));
|
||||
|
||||
return rows.map((text, i): TaskConfig => {
|
||||
const description = task.description ? `${task.description}\n${text}` : text;
|
||||
const isLast = i === rows.length - 1;
|
||||
const nextTasks = isLast ? inheritedNext : [names[i + 1]];
|
||||
const base: TaskConfig = {
|
||||
...inheritedExecution,
|
||||
description,
|
||||
name: names[i],
|
||||
agent: task.agent,
|
||||
expected_output: task.expected_output,
|
||||
onTaskComplete: task.onTaskComplete ?? task.callback,
|
||||
isStart: i === 0,
|
||||
taskType: options.decisionMode ? 'decision' : 'task',
|
||||
nextTasks,
|
||||
};
|
||||
if (options.decisionMode) {
|
||||
base.condition = {
|
||||
done: nextTasks.length > 0 ? nextTasks : [],
|
||||
retry: [names[i]],
|
||||
exit: [],
|
||||
};
|
||||
}
|
||||
return base;
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
/**
|
||||
* `loopOver` / `loopVar` / `loopState` fan-out.
|
||||
*
|
||||
* Port of the loop branch in
|
||||
* `praisonaiagents/workflows/workflows.py::_run_step_loop`: when a step names a
|
||||
* `loop_over` variable holding a list, the step runs once per item with
|
||||
* `all_variables[loop_var]` bound to that item and `_loop_index` to its
|
||||
* position. Here the fan-out is materialised as one child task per item so the
|
||||
* runner can execute them like any other tasks; each child records where it sat
|
||||
* in the loop on `loopState` (Python surfaces `loop_state` in the callback
|
||||
* metadata).
|
||||
*
|
||||
* Returns `TaskConfig` values rather than `Task` instances so the module never
|
||||
* imports the `Task` class at run time.
|
||||
*/
|
||||
|
||||
import type { Task, TaskConfig } from '../types';
|
||||
|
||||
/** Python's `_loop_index` key. */
|
||||
export const LOOP_INDEX_VAR = '_loop_index';
|
||||
|
||||
/**
|
||||
* The execution-affecting settings a fan-out child inherits from its parent, so
|
||||
* a task that combines `loopOver` / `inputFile` with a handler, output shaping,
|
||||
* retries, hooks, memory, knowledge or caching keeps those behaviours on every
|
||||
* iteration. Per-item fields (`description`, `name`, `nextTasks`, `isStart`,
|
||||
* `variables`, `loopState`, and — for the input-file path — `taskType` /
|
||||
* `condition`) are set by the caller and deliberately not copied here.
|
||||
*/
|
||||
export function inheritedExecutionConfig(task: Task): Partial<TaskConfig> {
|
||||
const inherited: Partial<TaskConfig> = {};
|
||||
if (task.tools.length > 0) inherited.tools = [...task.tools];
|
||||
if (task.images.length > 0) inherited.images = [...task.images];
|
||||
if (task.handler !== undefined) inherited.handler = task.handler;
|
||||
if (task.shouldRun !== undefined) inherited.shouldRun = task.shouldRun;
|
||||
if (task.agentConfig !== undefined) inherited.agentConfig = task.agentConfig;
|
||||
if (task.outputJson !== undefined) inherited.outputJson = task.outputJson;
|
||||
if (task.outputPydantic !== undefined) inherited.outputPydantic = task.outputPydantic;
|
||||
if (task.outputConfig !== undefined) inherited.outputConfig = task.outputConfig;
|
||||
if (task.output !== undefined) inherited.output = task.output;
|
||||
if (task.outputFile !== undefined) inherited.outputFile = task.outputFile;
|
||||
if (task.guardrails !== undefined) inherited.guardrails = task.guardrails;
|
||||
else if (task.guardrail !== undefined) inherited.guardrail = task.guardrail;
|
||||
if (task.memory !== undefined) inherited.memory = task.memory;
|
||||
if (task.knowledge !== undefined) inherited.knowledge = task.knowledge;
|
||||
if (task.hooks !== undefined) inherited.hooks = task.hooks;
|
||||
if (task.caching !== undefined) inherited.caching = task.caching;
|
||||
if (task.execution !== undefined) inherited.execution = task.execution;
|
||||
inherited.maxRetries = task.maxRetries;
|
||||
inherited.retryDelay = task.retryDelay;
|
||||
inherited.onError = task.onError;
|
||||
inherited.skipOnFailure = task.skipOnFailure;
|
||||
inherited.retainFullContext = task.retainFullContext;
|
||||
inherited.qualityCheck = task.qualityCheck;
|
||||
inherited.failOnCallbackError = task.failOnCallbackError;
|
||||
inherited.failOnMemoryError = task.failOnMemoryError;
|
||||
return inherited;
|
||||
}
|
||||
|
||||
function isIterable(value: unknown): value is unknown[] {
|
||||
return Array.isArray(value);
|
||||
}
|
||||
|
||||
function scalar(value: unknown): string | number {
|
||||
if (typeof value === 'number' || typeof value === 'string') return value;
|
||||
if (value === null || value === undefined) return '';
|
||||
if (typeof value === 'boolean') return value ? 'true' : 'false';
|
||||
try { return JSON.stringify(value); } catch { return String(value); }
|
||||
}
|
||||
|
||||
/**
|
||||
* Expand `task` into one config per item of `variables[task.loopOver]`.
|
||||
*
|
||||
* @returns `[]` when the task has no `loopOver`, when the named variable is
|
||||
* missing, or when it does not hold a list (Python logs
|
||||
* "loop_over variable '<name>' is not iterable" and runs the step once).
|
||||
*/
|
||||
export function loopTaskConfigs(task: Task, variables: Record<string, unknown> = {}): TaskConfig[] {
|
||||
if (!task.loopOver) return [];
|
||||
const scope = { ...task.variables, ...variables };
|
||||
const items = scope[task.loopOver];
|
||||
if (!isIterable(items)) return [];
|
||||
|
||||
const loopVar = task.loopVar || 'item';
|
||||
const inherited = inheritedExecutionConfig(task);
|
||||
return items.map((item, index): TaskConfig => ({
|
||||
...inherited,
|
||||
description: task.description,
|
||||
name: task.name ? `${task.name}_${index + 1}` : undefined,
|
||||
agent: task.agent,
|
||||
expected_output: task.expected_output,
|
||||
onTaskComplete: task.onTaskComplete ?? task.callback,
|
||||
variables: { ...task.variables, [loopVar]: item, [LOOP_INDEX_VAR]: index },
|
||||
loopState: {
|
||||
[loopVar]: scalar(item),
|
||||
index,
|
||||
total: items.length,
|
||||
},
|
||||
nextTasks: index === items.length - 1 && task.nextTasks.length > 0 ? [...task.nextTasks] : [],
|
||||
isStart: index === 0 && task.isStart,
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether `task` asks for a loop fan-out at all — true even when the variable
|
||||
* turns out to be missing, so a runner can tell "no loop" from "empty loop".
|
||||
*/
|
||||
export function isLoopTask(task: Task): boolean {
|
||||
return Boolean(task.loopOver);
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
/**
|
||||
* Per-task `memory` (and the `config.memory_config` that lazily creates it),
|
||||
* plus the `failOnMemoryError` policy.
|
||||
*
|
||||
* Ports `praisonaiagents/task/task.py::initialize_memory` / `store_in_memory`
|
||||
* and the memory-context branch of `agents.py::_prepare_task_prompt`:
|
||||
*
|
||||
* - `config.memory_config` builds a store on first use when the task was given
|
||||
* no `memory` object.
|
||||
* - A task with memory has its output stored after it runs, and recalls related
|
||||
* entries into the next prompt.
|
||||
* - A store that throws appends to `nonFatalErrors` and is otherwise swallowed,
|
||||
* unless `failOnMemoryError` is set, in which case it is re-raised.
|
||||
*/
|
||||
|
||||
import type { Task } from '../types';
|
||||
|
||||
/** The store shape both `Memory` and `Agent`'s memory satisfy. */
|
||||
export interface TaskMemoryStore {
|
||||
add(content: string, role: 'user' | 'assistant' | 'system', metadata?: Record<string, unknown>): Promise<unknown> | unknown;
|
||||
search(query: string, limit?: number): Promise<Array<{ entry: { content: string }; score: number }>>;
|
||||
}
|
||||
|
||||
/** Builds a store from `config.memory_config`; injectable so tests need no backend. */
|
||||
export type MemoryFactory = (config: Record<string, unknown>) => TaskMemoryStore;
|
||||
|
||||
function defaultMemoryFactory(config: Record<string, unknown>): TaskMemoryStore {
|
||||
// eslint-disable-next-line @typescript-eslint/no-var-requires
|
||||
const { Memory } = require('../../memory/memory') as { Memory: new (c: Record<string, unknown>) => TaskMemoryStore };
|
||||
return new Memory(config);
|
||||
}
|
||||
|
||||
function isStore(value: unknown): value is TaskMemoryStore {
|
||||
return typeof value === 'object' && value !== null
|
||||
&& typeof (value as TaskMemoryStore).add === 'function'
|
||||
&& typeof (value as TaskMemoryStore).search === 'function';
|
||||
}
|
||||
|
||||
/** The `memory_config` a task's `config` carries, in either casing. */
|
||||
export function memoryConfigOf(task: Task): Record<string, unknown> | undefined {
|
||||
const cfg = task.config;
|
||||
const value = cfg.memory_config ?? cfg.memoryConfig;
|
||||
return typeof value === 'object' && value !== null ? value as Record<string, unknown> : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* The task's store, creating one from `config.memory_config` on first use.
|
||||
* Assigns the result to `task.memory`, as Python does.
|
||||
*/
|
||||
export function initializeTaskMemory(task: Task, factory: MemoryFactory = defaultMemoryFactory): TaskMemoryStore | undefined {
|
||||
if (isStore(task.memory)) return task.memory;
|
||||
const config = memoryConfigOf(task);
|
||||
if (!config) return undefined;
|
||||
try {
|
||||
const store = factory(config);
|
||||
task.memory = store;
|
||||
return store;
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
task.nonFatalErrors.push(`initialize_memory: ${message}`);
|
||||
if (task.failOnMemoryError) throw err;
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Store `content` in the task's memory.
|
||||
*
|
||||
* A failure is recorded on `nonFatalErrors` and swallowed, unless
|
||||
* `failOnMemoryError` is set — that flag is the difference between a degraded
|
||||
* run and a failed one.
|
||||
*/
|
||||
export async function storeTaskOutput(
|
||||
task: Task,
|
||||
content: string,
|
||||
agentName = 'Agent',
|
||||
factory: MemoryFactory = defaultMemoryFactory
|
||||
): Promise<boolean> {
|
||||
const store = initializeTaskMemory(task, factory);
|
||||
if (!store) return false;
|
||||
try {
|
||||
await store.add(content, 'assistant', {
|
||||
agent_name: agentName,
|
||||
task_id: String(task.id),
|
||||
timestamp: Date.now() / 1000,
|
||||
});
|
||||
return true;
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
task.nonFatalErrors.push(`store_in_memory: ${message}`);
|
||||
if (task.failOnMemoryError) throw err;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Recall entries related to `query` and render them for the prompt.
|
||||
* Returns `''` when the task has no memory or nothing matched.
|
||||
*/
|
||||
export async function buildMemoryContext(
|
||||
task: Task,
|
||||
query: string,
|
||||
maxItems = 5,
|
||||
factory: MemoryFactory = defaultMemoryFactory
|
||||
): Promise<string> {
|
||||
const store = initializeTaskMemory(task, factory);
|
||||
if (!store) return '';
|
||||
try {
|
||||
const hits = await store.search(query, maxItems);
|
||||
if (!hits || hits.length === 0) return '';
|
||||
return hits.map((h) => h.entry.content).join('\n');
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
task.nonFatalErrors.push(`memory operations: ${message}`);
|
||||
if (task.failOnMemoryError) throw err;
|
||||
return '';
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
/**
|
||||
* Multimodal message assembly for `Task(images=[...])`.
|
||||
*
|
||||
* Port of `praisonaiagents/agents/agents.py::get_multimodal_message`: a local
|
||||
* file is inlined as a `data:` URI, anything else is passed through as a remote
|
||||
* image URL, and the text prompt leads the content list.
|
||||
*
|
||||
* One documented gap: Python's `.mp4` branch decodes the video with OpenCV and
|
||||
* appends one frame per second. There is no equivalent decoder in this package,
|
||||
* so a local video contributes a text note naming the file instead of silently
|
||||
* producing an unusable `data:video/...` URI. Pass extracted frames as images
|
||||
* to get the Python behaviour.
|
||||
*/
|
||||
|
||||
import * as fs from 'fs';
|
||||
import * as path from 'path';
|
||||
|
||||
/** One part of an OpenAI-style multimodal message. */
|
||||
export type MessageContentPart =
|
||||
| { type: 'text'; text: string }
|
||||
| { type: 'image_url'; image_url: { url: string } };
|
||||
|
||||
const VIDEO_EXTENSIONS: ReadonlySet<string> = new Set(['.mp4', '.mov', '.avi', '.mkv', '.webm']);
|
||||
|
||||
/**
|
||||
* Map an image extension to its registered media type. `.jpg` is `image/jpeg`
|
||||
* and `.svg` is `image/svg+xml`; a bare `image/<ext>` is not a registered type
|
||||
* and some providers reject the resulting data URI.
|
||||
*/
|
||||
const IMAGE_MIME_TYPES: Readonly<Record<string, string>> = {
|
||||
'.jpg': 'image/jpeg',
|
||||
'.jpeg': 'image/jpeg',
|
||||
'.png': 'image/png',
|
||||
'.gif': 'image/gif',
|
||||
'.webp': 'image/webp',
|
||||
'.bmp': 'image/bmp',
|
||||
'.svg': 'image/svg+xml',
|
||||
};
|
||||
|
||||
/** The filesystem calls this module needs; injectable so tests need no disk. */
|
||||
export interface ImageFileSystem {
|
||||
existsSync(p: string): boolean;
|
||||
readFileSync(p: string): Buffer;
|
||||
}
|
||||
|
||||
const realFs: ImageFileSystem = {
|
||||
existsSync: (p) => fs.existsSync(p),
|
||||
readFileSync: (p) => fs.readFileSync(p),
|
||||
};
|
||||
|
||||
/** The note used in place of Python's OpenCV frame extraction. */
|
||||
export function videoNote(file: string): string {
|
||||
return `[video: ${file}] frame extraction is not available in the TypeScript SDK; pass extracted frames as images instead.`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the message content for `text` plus `images`.
|
||||
*
|
||||
* With no images this is a single text part, which is what makes the option
|
||||
* observable: the same prompt with `images` gains one `image_url` part per
|
||||
* entry.
|
||||
*/
|
||||
export function buildMultimodalContent(
|
||||
text: string,
|
||||
images: readonly string[],
|
||||
fileSystem: ImageFileSystem = realFs
|
||||
): MessageContentPart[] {
|
||||
const content: MessageContentPart[] = [{ type: 'text', text }];
|
||||
for (const image of images) {
|
||||
if (!fileSystem.existsSync(image)) {
|
||||
content.push({ type: 'image_url', image_url: { url: image } });
|
||||
continue;
|
||||
}
|
||||
const ext = path.extname(image).toLowerCase();
|
||||
if (VIDEO_EXTENSIONS.has(ext)) {
|
||||
content.push({ type: 'text', text: videoNote(image) });
|
||||
continue;
|
||||
}
|
||||
const encoded = fileSystem.readFileSync(image).toString('base64');
|
||||
content.push({
|
||||
type: 'image_url',
|
||||
image_url: { url: `data:${IMAGE_MIME_TYPES[ext] ?? 'image/png'};base64,${encoded}` },
|
||||
});
|
||||
}
|
||||
return content;
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
/**
|
||||
* Turning an agent's raw text into a `TaskOutput`, and resolving the
|
||||
* `outputConfig` object onto the task's individual output fields.
|
||||
*
|
||||
* `buildTaskOutput` ports `praisonaiagents/agents/agents.py::_process_task_result`:
|
||||
* the raw text is stripped of a markdown fence and parsed when the task asked
|
||||
* for `outputJson` or `outputPydantic`; a parse failure is recorded as a
|
||||
* non-fatal error and the raw text stands.
|
||||
*
|
||||
* `resolveOutputConfig` ports
|
||||
* `praisonaiagents/workflows/workflow_configs.py::resolve_step_output_config`:
|
||||
* a string is a file path, an object supplies `file` / `json_model` /
|
||||
* `pydantic_model` / `variable`.
|
||||
*/
|
||||
|
||||
import type { Task, TaskOutput } from '../types';
|
||||
|
||||
/** The fields an `outputConfig` (Python `TaskOutputConfig`) can carry. */
|
||||
export interface ResolvedOutputConfig {
|
||||
file?: string;
|
||||
json?: unknown;
|
||||
pydantic?: unknown;
|
||||
variable?: string;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
/** Resolve an `outputConfig` value; `undefined` when it carries nothing. */
|
||||
export function resolveOutputConfig(value: unknown): ResolvedOutputConfig | undefined {
|
||||
if (value === undefined || value === null) return undefined;
|
||||
if (typeof value === 'string') return { file: value };
|
||||
if (!isRecord(value)) return undefined;
|
||||
const resolved: ResolvedOutputConfig = {};
|
||||
if (typeof value.file === 'string') resolved.file = value.file;
|
||||
const json = value.jsonModel ?? value.json_model ?? value.json;
|
||||
if (json !== undefined && json !== null) resolved.json = json;
|
||||
const pydantic = value.pydanticModel ?? value.pydantic_model ?? value.pydantic;
|
||||
if (pydantic !== undefined && pydantic !== null) resolved.pydantic = pydantic;
|
||||
if (typeof value.variable === 'string') resolved.variable = value.variable;
|
||||
return Object.keys(resolved).length > 0 ? resolved : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Python `Agents.clean_json_output`: strip a ```json fence so the payload can
|
||||
* be parsed.
|
||||
*/
|
||||
export function cleanJsonOutput(raw: string): string {
|
||||
let text = raw.trim();
|
||||
if (text.startsWith('```')) {
|
||||
text = text.replace(/^```[a-zA-Z]*\s*/, '');
|
||||
if (text.endsWith('```')) text = text.slice(0, -3);
|
||||
}
|
||||
return text.trim();
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the `TaskOutput` for `raw`.
|
||||
*
|
||||
* A task with neither `outputJson` nor `outputPydantic` gets a plain `RAW`
|
||||
* output; asking for either parses the text and sets the matching field and
|
||||
* `outputFormat`. A parse failure is appended to the task's `nonFatalErrors`.
|
||||
*/
|
||||
export function buildTaskOutput(task: Task, raw: string, agentName = 'Agent'): TaskOutput {
|
||||
const output: TaskOutput = {
|
||||
description: task.description,
|
||||
summary: task.description.slice(0, 10),
|
||||
raw,
|
||||
agent: agentName,
|
||||
outputFormat: 'RAW',
|
||||
};
|
||||
if (task.outputJson === undefined && task.outputPydantic === undefined) return output;
|
||||
|
||||
const cleaned = cleanJsonOutput(raw);
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(cleaned);
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
task.nonFatalErrors.push(`output parse: ${message}`);
|
||||
return output;
|
||||
}
|
||||
|
||||
if (task.outputJson !== undefined) {
|
||||
if (isRecord(parsed)) {
|
||||
output.outputJson = parsed;
|
||||
output.outputFormat = 'JSON';
|
||||
} else {
|
||||
task.nonFatalErrors.push('output parse: expected a JSON object');
|
||||
}
|
||||
}
|
||||
if (task.outputPydantic !== undefined) {
|
||||
if (isRecord(parsed)) {
|
||||
output.outputPydantic = parsed;
|
||||
output.outputFormat = 'Pydantic';
|
||||
} else {
|
||||
task.nonFatalErrors.push('output parse: expected a JSON object for pydantic');
|
||||
}
|
||||
}
|
||||
return output;
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
/**
|
||||
* Retry pacing and failure policy for a Task.
|
||||
*
|
||||
* Ports three Python behaviours that live in the loop that *runs* tasks, not
|
||||
* in `Task.__init__`:
|
||||
*
|
||||
* - `retry_delay` — `praisonaiagents/agents/agents.py` waits
|
||||
* `min(retry_delay * 2 ** attempt, max_retry_delay)` seconds between
|
||||
* attempts, so the delay is exponential and capped at five minutes.
|
||||
* - `skip_on_failure` / `on_error` — `Agents._should_continue_on_dep_failure`
|
||||
* lets a task run in degraded mode when an upstream dependency failed
|
||||
* instead of being force-failed with it.
|
||||
* - `rerun` — whether an already-completed task is executed again when the
|
||||
* workflow reaches it a second time (`TaskExecutionConfig.rerun`).
|
||||
*/
|
||||
|
||||
import type { Task } from '../types';
|
||||
|
||||
/** Python `max_retry_delay`: the five-minute cap on exponential backoff. */
|
||||
export const MAX_RETRY_DELAY_SECONDS = 300;
|
||||
|
||||
/** The fields the retry policy reads; a plain object works as well as a Task. */
|
||||
export type RetryPolicyLike = Pick<Task, 'retryDelay' | 'onError' | 'skipOnFailure' | 'rerun' | 'status'> & {
|
||||
maxRetryDelay?: number;
|
||||
};
|
||||
|
||||
/**
|
||||
* Seconds to wait before retry number `attempt` (0-based), following Python's
|
||||
* `min(delay * (2 ** retries), max_delay)`. A task with `retryDelay: 0` (the
|
||||
* default) never waits, which is what makes the option observable.
|
||||
*/
|
||||
export function retryDelaySeconds(task: RetryPolicyLike, attempt: number): number {
|
||||
const base = Number.isFinite(task.retryDelay) ? Math.max(0, task.retryDelay) : 0;
|
||||
if (base === 0) return 0;
|
||||
const cap = task.maxRetryDelay ?? MAX_RETRY_DELAY_SECONDS;
|
||||
const exponent = Math.max(0, Math.floor(attempt));
|
||||
return Math.min(base * Math.pow(2, exponent), cap);
|
||||
}
|
||||
|
||||
/** Sleep helper; `timer` is injectable so tests never wait on a real clock. */
|
||||
export function sleepSeconds(
|
||||
seconds: number,
|
||||
timer: (fn: () => void, ms: number) => unknown = setTimeout
|
||||
): Promise<void> {
|
||||
if (seconds <= 0) return Promise.resolve();
|
||||
return new Promise((resolve) => { timer(resolve, seconds * 1000); });
|
||||
}
|
||||
|
||||
/**
|
||||
* Python `Agents._should_continue_on_dep_failure`: true when the task opted
|
||||
* into graceful degradation, so an upstream failure does not force-fail it.
|
||||
*/
|
||||
export function continuesOnDependencyFailure(task: RetryPolicyLike): boolean {
|
||||
return Boolean(task.skipOnFailure) || task.onError === 'continue';
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether the runner should execute this task now. A task that already
|
||||
* completed is skipped unless `rerun` was requested
|
||||
* (Python `TaskExecutionConfig.rerun`).
|
||||
*/
|
||||
export function needsRun(task: RetryPolicyLike): boolean {
|
||||
if (task.status !== 'completed') return true;
|
||||
return Boolean(task.rerun);
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
/**
|
||||
* Workflow routing for a Task: which task runs first, and which runs next.
|
||||
*
|
||||
* Port of `praisonaiagents/process/process.py::workflow()`. The precedence it
|
||||
* applies, in order:
|
||||
*
|
||||
* 1. A `decision`/`loop` task consults its `condition` (alias `routing`) table
|
||||
* with the decision the model returned; an empty target, or the literal
|
||||
* `"exit"`, ends the workflow.
|
||||
* 2. Otherwise the unified `when` / `thenTask` / `elseTask` routing wins over
|
||||
* `nextTasks` when configured (`_resolve_when_routing`).
|
||||
* 3. Otherwise `nextTasks[0]`.
|
||||
*
|
||||
* `isStart` picks the entry point and `nextTasks` defines the edges that
|
||||
* become each task's `previousTasks`.
|
||||
*/
|
||||
|
||||
import type { Task } from '../types';
|
||||
import { evaluateWhen } from './conditions';
|
||||
|
||||
/** Sentinel returned when routing says the workflow is finished. */
|
||||
export const EXIT = 'exit' as const;
|
||||
|
||||
/** Where a task sends control next. */
|
||||
export type RouteDecision =
|
||||
| { kind: 'task'; name: string }
|
||||
| { kind: 'exit' }
|
||||
| { kind: 'none' };
|
||||
|
||||
/** Variables a `when` expression and a decision table are evaluated against. */
|
||||
export interface RoutingContext extends Record<string, unknown> {
|
||||
/** Raw text the task produced. */
|
||||
previous_output?: string;
|
||||
/** Structured `decision` field, for `decision`/`loop` task types. */
|
||||
decision?: string;
|
||||
}
|
||||
|
||||
/** True when the task uses the unified when/then/else routing (`_has_when_routing`). */
|
||||
export function hasWhenRouting(task: Task): boolean {
|
||||
return task.when !== undefined || task.thenTask !== undefined || task.elseTask !== undefined;
|
||||
}
|
||||
|
||||
/** Evaluate the task's `when` expression; a task without one always passes. */
|
||||
export function evaluateTaskWhen(task: Task, context: RoutingContext = {}): boolean {
|
||||
if (task.when === undefined) return true;
|
||||
return evaluateWhen(task.when, context);
|
||||
}
|
||||
|
||||
/**
|
||||
* Look a decision up in the task's `condition`/`routing` table.
|
||||
* Returns `exit` for a missing, empty or literally-"exit" target.
|
||||
*/
|
||||
export function decisionRoute(task: Task, decision: string): RouteDecision {
|
||||
const table = task.condition;
|
||||
if (!table || Object.keys(table).length === 0) return { kind: 'none' };
|
||||
const targets = table[decision.toLowerCase().trim()] ?? table[decision];
|
||||
if (targets === undefined) return { kind: 'exit' };
|
||||
const list = Array.isArray(targets) ? targets : [targets];
|
||||
if (list.length === 0 || list[0] === EXIT) return { kind: 'exit' };
|
||||
return { kind: 'task', name: String(list[0]) };
|
||||
}
|
||||
|
||||
/**
|
||||
* The task control moves to after this one finished with `context`.
|
||||
*
|
||||
* `context.decision` (or, failing that, the raw output) drives the decision
|
||||
* table for `decision`/`loop` tasks; `when`/`thenTask`/`elseTask` and finally
|
||||
* `nextTasks[0]` are consulted after it.
|
||||
*/
|
||||
export function resolveNextTask(task: Task, context: RoutingContext = {}): RouteDecision {
|
||||
if (task.taskType === 'decision' || task.taskType === 'loop') {
|
||||
const decision = String(context.decision ?? context.previous_output ?? '').toLowerCase().trim();
|
||||
if (decision.length > 0) {
|
||||
const routed = decisionRoute(task, decision);
|
||||
if (routed.kind !== 'none') return routed;
|
||||
}
|
||||
}
|
||||
|
||||
if (hasWhenRouting(task)) {
|
||||
const target = evaluateTaskWhen(task, context) ? task.thenTask : task.elseTask;
|
||||
if (target !== undefined) return { kind: 'task', name: target };
|
||||
// when-routing is authoritative: with no `nextTasks` fallback the path
|
||||
// ends here rather than picking an unrelated task.
|
||||
if (task.nextTasks.length === 0) return { kind: 'none' };
|
||||
}
|
||||
|
||||
if (task.nextTasks.length > 0) return { kind: 'task', name: task.nextTasks[0] };
|
||||
return { kind: 'none' };
|
||||
}
|
||||
|
||||
/**
|
||||
* The task a workflow starts at: the one flagged `isStart`, else the first
|
||||
* task with no incoming `nextTasks` edge, else the first task
|
||||
* (Python `process.workflow()`'s start-task selection).
|
||||
*/
|
||||
export function startTaskOf(tasks: readonly Task[]): Task | undefined {
|
||||
if (tasks.length === 0) return undefined;
|
||||
const flagged = tasks.find((t) => t.isStart);
|
||||
if (flagged) return flagged;
|
||||
const targeted = new Set<string>();
|
||||
for (const t of tasks) for (const n of t.nextTasks) targeted.add(n);
|
||||
const unreferenced = tasks.find((t) => t.name === undefined || !targeted.has(t.name));
|
||||
return unreferenced ?? tasks[0];
|
||||
}
|
||||
|
||||
/**
|
||||
* Fill each task's `previousTasks` from the `nextTasks` edges of the others
|
||||
* (Python does this at the top of `process.workflow()`). Returns the map it
|
||||
* wrote, keyed by task name.
|
||||
*/
|
||||
export function linkPreviousTasks(tasks: readonly Task[]): Map<string, string[]> {
|
||||
const byName = new Map<string, Task>();
|
||||
for (const t of tasks) if (t.name) byName.set(t.name, t);
|
||||
|
||||
for (const t of tasks) t.previousTasks = [];
|
||||
for (const t of tasks) {
|
||||
if (!t.name) continue;
|
||||
for (const next of t.nextTasks) {
|
||||
const target = byName.get(next);
|
||||
if (target && !target.previousTasks.includes(t.name)) target.previousTasks.push(t.name);
|
||||
}
|
||||
}
|
||||
|
||||
const map = new Map<string, string[]>();
|
||||
for (const [name, t] of byName) map.set(name, [...t.previousTasks]);
|
||||
return map;
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
/**
|
||||
* Batching of tasks for `asyncExecution` (Python `async_execution`).
|
||||
*
|
||||
* Port of the loop in `praisonaiagents/agents/agents.py::arun_all_tasks`:
|
||||
* consecutive tasks marked `async_execution` are collected and gathered
|
||||
* together, and the batch is flushed before any synchronous task runs, so a
|
||||
* sync task never overlaps the async ones queued ahead of it. A task whose
|
||||
* dependency is still pending in the open batch forces an early flush, because
|
||||
* otherwise it would build its prompt from an empty dependency result
|
||||
* (`Agents._depends_on_pending`).
|
||||
*/
|
||||
|
||||
import type { Task } from '../types';
|
||||
|
||||
/** The fields batching reads; a plain object works as well as a Task. */
|
||||
export interface SchedulableTask {
|
||||
asyncExecution: boolean;
|
||||
/** Tasks whose results this one consumes (`dependencies` on a real Task). */
|
||||
dependencies?: readonly unknown[];
|
||||
}
|
||||
|
||||
/** One step of the run plan: either a parallel fan-out or a single serial task. */
|
||||
export interface TaskBatch<T> {
|
||||
/** True when every task in `tasks` may run concurrently. */
|
||||
parallel: boolean;
|
||||
tasks: T[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Group `tasks` into the batches a runner should execute in order.
|
||||
*
|
||||
* Without `asyncExecution` every task is its own serial batch (Python's
|
||||
* default). With it, runs of async tasks collapse into one parallel batch.
|
||||
*/
|
||||
export function planTaskBatches<T extends SchedulableTask>(tasks: readonly T[]): Array<TaskBatch<T>> {
|
||||
const batches: Array<TaskBatch<T>> = [];
|
||||
let open: T[] = [];
|
||||
|
||||
const flush = (): void => {
|
||||
if (open.length === 0) return;
|
||||
batches.push({ parallel: open.length > 1, tasks: open });
|
||||
open = [];
|
||||
};
|
||||
|
||||
for (const task of tasks) {
|
||||
if (!task.asyncExecution) {
|
||||
flush();
|
||||
batches.push({ parallel: false, tasks: [task] });
|
||||
continue;
|
||||
}
|
||||
// A dependency still queued in the open batch has no result yet, so the
|
||||
// batch must be flushed before this task joins one (Python
|
||||
// `_depends_on_pending`).
|
||||
if (dependsOnPending(task, open)) flush();
|
||||
open.push(task);
|
||||
}
|
||||
flush();
|
||||
return batches;
|
||||
}
|
||||
|
||||
/** True when `task` depends on one of the tasks already queued in `pending`. */
|
||||
export function dependsOnPending<T extends SchedulableTask>(task: T, pending: readonly T[]): boolean {
|
||||
const deps = task.dependencies;
|
||||
if (!deps || deps.length === 0 || pending.length === 0) return false;
|
||||
return deps.some((dep) => pending.some((queued) => queued === (dep as unknown as T)));
|
||||
}
|
||||
|
||||
/** Narrowing helper so a `Task[]` can be planned without a cast. */
|
||||
export function planTaskRun(tasks: readonly Task[]): Array<TaskBatch<Task>> {
|
||||
return planTaskBatches(tasks as unknown as readonly (Task & SchedulableTask)[]) as Array<TaskBatch<Task>>;
|
||||
}
|
||||
@@ -4,6 +4,35 @@ import { OpenAIService } from '../llm/openai';
|
||||
import { Logger } from '../utils/logger';
|
||||
import { randomUUID } from '../utils/uuid';
|
||||
import { notYetHonoured, unhonouredFor } from '../utils/parity-notice';
|
||||
import {
|
||||
continuesOnDependencyFailure,
|
||||
needsRun,
|
||||
retryDelaySeconds,
|
||||
sleepSeconds,
|
||||
} from './engine/task-retry';
|
||||
import { buildTaskContext, type PreviousOutput } from './engine/task-context';
|
||||
import {
|
||||
evaluateTaskWhen,
|
||||
hasWhenRouting,
|
||||
resolveNextTask,
|
||||
type RouteDecision,
|
||||
type RoutingContext,
|
||||
} from './engine/task-routing';
|
||||
import { inputFileTaskConfigs, type FileReader } from './engine/task-input-file';
|
||||
import { loopTaskConfigs } from './engine/task-loop';
|
||||
import { buildMultimodalContent, type ImageFileSystem, type MessageContentPart } from './engine/task-messages';
|
||||
import { buildTaskOutput, resolveOutputConfig } from './engine/task-output';
|
||||
import { runTaskHandler, type HandlerContext, type HandlerResult } from './engine/task-handler';
|
||||
import { resolveTaskAgent, type ResolveAgentOptions } from './engine/task-agent';
|
||||
import {
|
||||
buildMemoryContext,
|
||||
initializeTaskMemory,
|
||||
storeTaskOutput,
|
||||
type MemoryFactory,
|
||||
type TaskMemoryStore,
|
||||
} from './engine/task-memory';
|
||||
import { buildKnowledgeContext, resolveTaskCache, resolveTaskHooks } from './engine/task-features';
|
||||
import type { ResponseCache, TaskHooks } from './engine/types';
|
||||
|
||||
/**
|
||||
* Result of running a task. Mirrors Python `praisonaiagents.output.models.TaskOutput`.
|
||||
@@ -29,8 +58,30 @@ export interface TaskOutput {
|
||||
nonFatalErrors?: string[];
|
||||
}
|
||||
|
||||
/** A task completion callback; sync or async. */
|
||||
export type TaskCallback = (output: TaskOutput) => unknown | Promise<unknown>;
|
||||
/**
|
||||
* Python parity: the metadata dict `Task._execute_callback_with_metadata`
|
||||
* hands to a callback that declares a second parameter. It is the only place
|
||||
* `loopState`, `inputFile`, `taskType` and `asyncExecution` reach user code.
|
||||
*/
|
||||
export interface TaskCallbackMetadata {
|
||||
taskId: number | string;
|
||||
taskName?: string;
|
||||
agentName?: string;
|
||||
taskType: string;
|
||||
taskStatus: string;
|
||||
taskDescription: string;
|
||||
expectedOutput: string;
|
||||
inputFile?: string;
|
||||
loopState: Record<string, string | number>;
|
||||
retryCount: number;
|
||||
asyncExecution: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* A task completion callback; sync or async. A callback that declares a second
|
||||
* parameter also receives {@link TaskCallbackMetadata}, as in Python.
|
||||
*/
|
||||
export type TaskCallback = (output: TaskOutput, metadata?: TaskCallbackMetadata) => unknown | Promise<unknown>;
|
||||
|
||||
/**
|
||||
* A guardrail: either a validator returning `[ok, payload]` (sync or async)
|
||||
@@ -318,6 +369,12 @@ export class Task {
|
||||
nonFatalErrors: string[];
|
||||
/** Python parity: validation_feedback — last guardrail failure, fed back on retry. */
|
||||
validationFeedback?: unknown;
|
||||
/** Python parity: previous_tasks — names of the tasks whose `nextTasks` lead here. */
|
||||
previousTasks: string[];
|
||||
/** `hooks` resolved to the callbacks fired around this task, when it named any. */
|
||||
resolvedHooks?: TaskHooks;
|
||||
/** `caching` resolved to a per-task response cache, when it is enabled. */
|
||||
cache?: ResponseCache;
|
||||
|
||||
constructor(config: TaskConfig) {
|
||||
// `action` is the user-friendly alias for `description` (Python does the same).
|
||||
@@ -394,8 +451,21 @@ export class Task {
|
||||
this.failOnCallbackError = config.failOnCallbackError ?? false;
|
||||
this.failOnMemoryError = config.failOnMemoryError ?? false;
|
||||
this.nonFatalErrors = [];
|
||||
this.previousTasks = [];
|
||||
this.resolvedHooks = resolveTaskHooks(config.hooks);
|
||||
this.cache = resolveTaskCache(config.caching);
|
||||
|
||||
this.resolveExecutionConfig(config);
|
||||
// `outputConfig` fills the same fields as the unified `output` param but
|
||||
// only where the individual param was not supplied; `output` is applied
|
||||
// afterwards and wins over both (Python's stated precedence).
|
||||
const fromOutputConfig = resolveOutputConfig(config.outputConfig);
|
||||
if (fromOutputConfig) {
|
||||
if (fromOutputConfig.file !== undefined && config.outputFile === undefined) this.outputFile = fromOutputConfig.file;
|
||||
if (fromOutputConfig.json !== undefined && config.outputJson === undefined) this.outputJson = fromOutputConfig.json;
|
||||
if (fromOutputConfig.pydantic !== undefined && config.outputPydantic === undefined) this.outputPydantic = fromOutputConfig.pydantic;
|
||||
if (fromOutputConfig.variable !== undefined && config.outputVariable === undefined) this.outputVariable = fromOutputConfig.variable;
|
||||
}
|
||||
this.resolveUnifiedOutput(config);
|
||||
|
||||
if (this.outputJson !== undefined && this.outputPydantic !== undefined) {
|
||||
@@ -512,6 +582,183 @@ export class Task {
|
||||
return this.retryCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `retry_delay`. Seconds to wait before retry `attempt`
|
||||
* (0-based), doubling each time and capped at five minutes. A task with the
|
||||
* default `retryDelay` of 0 never waits.
|
||||
*/
|
||||
retryDelayFor(attempt: number): number {
|
||||
return retryDelaySeconds(this, attempt);
|
||||
}
|
||||
|
||||
/** Sleep `retryDelayFor(attempt)` seconds. `timer` is injectable for tests. */
|
||||
async waitBeforeRetry(attempt: number, timer?: (fn: () => void, ms: number) => unknown): Promise<number> {
|
||||
const seconds = this.retryDelayFor(attempt);
|
||||
await sleepSeconds(seconds, timer);
|
||||
return seconds;
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `skip_on_failure` / `on_error`
|
||||
* (`Agents._should_continue_on_dep_failure`). True when this task still runs,
|
||||
* in degraded mode, after an upstream dependency failed.
|
||||
*/
|
||||
continuesOnDependencyFailure(): boolean {
|
||||
return continuesOnDependencyFailure(this);
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `rerun`. Whether the runner should execute this task now —
|
||||
* false for a task that already completed unless `rerun` was requested.
|
||||
*/
|
||||
needsRun(): boolean {
|
||||
return needsRun(this);
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ context
|
||||
|
||||
/**
|
||||
* Python parity: `retain_full_context` (`Process._build_task_context`).
|
||||
* Assemble the upstream context block for this task: every previous result
|
||||
* when `retainFullContext` is set, only the most recent one otherwise.
|
||||
* Consumes `validationFeedback`, so it is injected exactly once.
|
||||
*/
|
||||
buildContext(previous: readonly PreviousOutput[] = []): string {
|
||||
return buildTaskContext(this, previous);
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ routing
|
||||
|
||||
/** True when this task uses the unified `when`/`thenTask`/`elseTask` routing. */
|
||||
hasRouting(): boolean {
|
||||
return hasWhenRouting(this) || this.nextTasks.length > 0 || Object.keys(this.condition).length > 0;
|
||||
}
|
||||
|
||||
/** Python parity: `evaluate_when`. A task without a `when` always passes. */
|
||||
evaluateWhen(context: RoutingContext = {}): boolean {
|
||||
return evaluateTaskWhen(this, context);
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `get_next_task` plus the `condition`/`routing` and
|
||||
* `nextTasks` fallbacks the workflow engine applies. Returns the routing
|
||||
* decision: a named task, `exit`, or nothing.
|
||||
*/
|
||||
routeAfter(context: RoutingContext = {}): RouteDecision {
|
||||
return resolveNextTask(this, context);
|
||||
}
|
||||
|
||||
/** The name of the task to run next, or `undefined` for exit / no route. */
|
||||
nextTaskFor(context: RoutingContext = {}): string | undefined {
|
||||
const route = this.routeAfter(context);
|
||||
return route.kind === 'task' ? route.name : undefined;
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ fan-out
|
||||
|
||||
/**
|
||||
* Python parity: `loop_over` / `loop_var` / `loop_state`
|
||||
* (`Workflows._run_step_loop`). One child task per item of the named
|
||||
* variable, with `loopVar` bound in the child's `variables` and the
|
||||
* position recorded on its `loopState`. `[]` when there is no loop.
|
||||
*/
|
||||
expandLoop(variables: Record<string, unknown> = {}): Task[] {
|
||||
return loopTaskConfigs(this, variables).map((config) => new Task(config));
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `input_file` (`Process._create_loop_subtasks`). One child
|
||||
* task per CSV row or text line, chained with `nextTasks`, the first flagged
|
||||
* `isStart` and the last inheriting this task's `nextTasks`. `[]` when the
|
||||
* task has no `inputFile`.
|
||||
*/
|
||||
expandFromInputFile(options: { decisionMode?: boolean; readFile?: FileReader } = {}): Task[] {
|
||||
return inputFileTaskConfigs(this, options).map((config) => new Task(config));
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ execution
|
||||
|
||||
/**
|
||||
* Python parity: `handler` (`Workflows._run_step`). Run the task's own
|
||||
* function instead of an agent. `undefined` when there is no handler, so the
|
||||
* caller falls through to the agent path.
|
||||
*/
|
||||
runHandler(context: HandlerContext): Promise<HandlerResult | undefined> {
|
||||
return runTaskHandler(this, context);
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `agent_config` (`Agents._create_agent_from_config`). The
|
||||
* agent that executes this task; a constructed one is assigned to `agent`.
|
||||
*/
|
||||
resolveAgent(options: ResolveAgentOptions = {}): unknown | undefined {
|
||||
return resolveTaskAgent(this, options);
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `images` (`get_multimodal_message`). The message content for
|
||||
* `prompt`: plain text without images, a content list with one `image_url`
|
||||
* part per image with them.
|
||||
*/
|
||||
buildMessageContent(prompt: string, fileSystem?: ImageFileSystem): string | MessageContentPart[] {
|
||||
if (this.images.length === 0) return prompt;
|
||||
return buildMultimodalContent(prompt, this.images, fileSystem);
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `_process_task_result`. Wrap `raw` in a TaskOutput, parsing
|
||||
* it when the task asked for `outputJson` or `outputPydantic`.
|
||||
*/
|
||||
buildOutput(raw: string, agentName = 'Agent'): TaskOutput {
|
||||
return buildTaskOutput(this, raw, agentName);
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ features
|
||||
|
||||
/** Python parity: `config` — the verbosity level a `config.verbose` requested. */
|
||||
get verboseLevel(): number {
|
||||
const value = this.config.verbose;
|
||||
return typeof value === 'number' ? value : 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `memory` plus `config.memory_config` (`initialize_memory`).
|
||||
* The task's store, created from `config.memory_config` on first use.
|
||||
*/
|
||||
initializeMemory(factory?: MemoryFactory): TaskMemoryStore | undefined {
|
||||
return initializeTaskMemory(this, factory);
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `store_in_memory`. Store `content` in the task's memory.
|
||||
* A failure lands on `nonFatalErrors` and is swallowed unless
|
||||
* `failOnMemoryError` is set. Returns whether anything was stored.
|
||||
*/
|
||||
storeInMemory(content: string, agentName = 'Agent', factory?: MemoryFactory): Promise<boolean> {
|
||||
return storeTaskOutput(this, content, agentName, factory);
|
||||
}
|
||||
|
||||
/** Python parity: `memory` — recall related entries for the next prompt. */
|
||||
memoryContext(query: string, maxItems = 5, factory?: MemoryFactory): Promise<string> {
|
||||
return buildMemoryContext(this, query, maxItems, factory);
|
||||
}
|
||||
|
||||
/** Python parity: `knowledge` — the knowledge block for this task's prompt. */
|
||||
knowledgeContext(query: string, limit = 5): Promise<string> {
|
||||
return buildKnowledgeContext(this, query, limit);
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: `hooks` — fire this task's `onTaskStart` hook. A task
|
||||
* without hooks does nothing.
|
||||
*/
|
||||
async notifyStart(index = 0): Promise<void> {
|
||||
// Idempotent: a runner that already called markStarted() does not get a
|
||||
// spurious "invalid transition" warning for calling this too.
|
||||
if (this.status !== TASK_STATUS.IN_PROGRESS) this.markStarted();
|
||||
if (this.resolvedHooks?.onTaskStart) await this.resolvedHooks.onTaskStart(this, index);
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ output
|
||||
|
||||
/** The workflow variable this task's output is stored under: `outputVariable`, else `name`. */
|
||||
@@ -564,10 +811,15 @@ export class Task {
|
||||
* `failOnCallbackError` is set.
|
||||
*/
|
||||
async notifyComplete(output: TaskOutput): Promise<TaskOutput> {
|
||||
const metadata = this.callbackMetadata();
|
||||
for (const cb of [this.callback, this.onTaskComplete]) {
|
||||
if (!cb) continue;
|
||||
try {
|
||||
await cb(output);
|
||||
// Metadata is always passed as a second argument; JavaScript
|
||||
// callbacks declaring fewer parameters (including a defaulted
|
||||
// `metadata = {}`) safely ignore the extra argument, so a
|
||||
// one-parameter callback still works.
|
||||
await cb(output, metadata);
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
this.nonFatalErrors.push(`callback: ${message}`);
|
||||
@@ -579,12 +831,48 @@ export class Task {
|
||||
}
|
||||
}
|
||||
}
|
||||
// Python parity: `hooks` — the per-task completion hook fires after the
|
||||
// user callbacks, with the same non-fatal error policy.
|
||||
const hook = this.resolvedHooks?.onTaskComplete;
|
||||
if (hook) {
|
||||
try {
|
||||
await hook(this, output);
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
this.nonFatalErrors.push(`hooks.onTaskComplete: ${message}`);
|
||||
Logger.error(`Task ${this.id}: Failed to execute onTaskComplete hook: ${message}`);
|
||||
if (this.failOnCallbackError) {
|
||||
output.nonFatalErrors = [...this.nonFatalErrors];
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (this.nonFatalErrors.length > 0 && output.nonFatalErrors === undefined) {
|
||||
output.nonFatalErrors = [...this.nonFatalErrors];
|
||||
}
|
||||
return output;
|
||||
}
|
||||
|
||||
/**
|
||||
* Python parity: the metadata dict handed to a two-parameter callback
|
||||
* (`_execute_callback_with_metadata`).
|
||||
*/
|
||||
callbackMetadata(): TaskCallbackMetadata {
|
||||
return {
|
||||
taskId: this.id,
|
||||
taskName: this.name,
|
||||
agentName: this.agent?.name,
|
||||
taskType: this.taskType,
|
||||
taskStatus: this.status,
|
||||
taskDescription: this.description,
|
||||
expectedOutput: this.expected_output,
|
||||
inputFile: this.inputFile,
|
||||
loopState: this.loopState,
|
||||
retryCount: this.retryCount,
|
||||
asyncExecution: this.asyncExecution,
|
||||
};
|
||||
}
|
||||
|
||||
toString(): string {
|
||||
const agentName = this.agent?.name ?? 'None';
|
||||
return `Task(name='${this.name ?? 'None'}', description='${this.description}', agent='${agentName}', status='${this.status}')`;
|
||||
|
||||
@@ -0,0 +1,676 @@
|
||||
/**
|
||||
* Behaviour tests for the Task execution-engine modules.
|
||||
*
|
||||
* Each block pairs an option with a control showing the behaviour is absent
|
||||
* when the option is not passed. No network, no disk unless the test writes it.
|
||||
*/
|
||||
process.env.PRAISONAI_PARITY_SILENT = '1';
|
||||
|
||||
import { Task } from '../../../../src/agent/types';
|
||||
import {
|
||||
MAX_RETRY_DELAY_SECONDS,
|
||||
continuesOnDependencyFailure,
|
||||
needsRun,
|
||||
retryDelaySeconds,
|
||||
sleepSeconds,
|
||||
} from '../../../../src/agent/engine/task-retry';
|
||||
import { dependsOnPending, planTaskBatches, planTaskRun } from '../../../../src/agent/engine/task-schedule';
|
||||
import { buildTaskContext, contextOutputs, renderValidationFeedback } from '../../../../src/agent/engine/task-context';
|
||||
import {
|
||||
decisionRoute,
|
||||
linkPreviousTasks,
|
||||
resolveNextTask,
|
||||
startTaskOf,
|
||||
} from '../../../../src/agent/engine/task-routing';
|
||||
import { inputFileRows, inputFileTaskConfigs, parseCsvLine } from '../../../../src/agent/engine/task-input-file';
|
||||
import { loopTaskConfigs } from '../../../../src/agent/engine/task-loop';
|
||||
import { buildMultimodalContent, videoNote } from '../../../../src/agent/engine/task-messages';
|
||||
import { buildTaskOutput, cleanJsonOutput, resolveOutputConfig } from '../../../../src/agent/engine/task-output';
|
||||
import { runTaskHandler } from '../../../../src/agent/engine/task-handler';
|
||||
import { agentOptionsFor, resolveTaskAgent } from '../../../../src/agent/engine/task-agent';
|
||||
import { buildMemoryContext, memoryConfigOf, storeTaskOutput } from '../../../../src/agent/engine/task-memory';
|
||||
import {
|
||||
buildKnowledgeContext,
|
||||
cacheKey,
|
||||
resolveTaskCache,
|
||||
resolveTaskHooks,
|
||||
} from '../../../../src/agent/engine/task-features';
|
||||
|
||||
// --------------------------------------------------------------- retry
|
||||
|
||||
describe('engine/task-retry: retryDelay, skipOnFailure, rerun', () => {
|
||||
it('retryDelay drives exponential backoff; the default of 0 never waits', () => {
|
||||
const withDelay = new Task({ description: 'd', retryDelay: 2 });
|
||||
expect(retryDelaySeconds(withDelay, 0)).toBe(2);
|
||||
expect(retryDelaySeconds(withDelay, 1)).toBe(4);
|
||||
expect(retryDelaySeconds(withDelay, 3)).toBe(16);
|
||||
|
||||
// Control: no retryDelay means no wait at any attempt.
|
||||
const control = new Task({ description: 'd' });
|
||||
expect(retryDelaySeconds(control, 0)).toBe(0);
|
||||
expect(retryDelaySeconds(control, 5)).toBe(0);
|
||||
});
|
||||
|
||||
it('caps the backoff at the Python five-minute ceiling', () => {
|
||||
const t = new Task({ description: 'd', retryDelay: 10 });
|
||||
expect(retryDelaySeconds(t, 30)).toBe(MAX_RETRY_DELAY_SECONDS);
|
||||
});
|
||||
|
||||
it('sleepSeconds schedules the computed wait and skips a zero one', async () => {
|
||||
const scheduled: number[] = [];
|
||||
const timer = (fn: () => void, ms: number): unknown => { scheduled.push(ms); fn(); return 0; };
|
||||
await sleepSeconds(retryDelaySeconds(new Task({ description: 'd', retryDelay: 1.5 }), 1), timer);
|
||||
await sleepSeconds(retryDelaySeconds(new Task({ description: 'd' }), 1), timer);
|
||||
expect(scheduled).toEqual([3000]);
|
||||
});
|
||||
|
||||
it('skipOnFailure (or onError=continue) keeps a task running after a dependency fails', () => {
|
||||
expect(continuesOnDependencyFailure(new Task({ description: 'd', skipOnFailure: true }))).toBe(true);
|
||||
expect(continuesOnDependencyFailure(new Task({ description: 'd', onError: 'continue' }))).toBe(true);
|
||||
// Control: the default force-fails with the dependency.
|
||||
expect(continuesOnDependencyFailure(new Task({ description: 'd' }))).toBe(false);
|
||||
});
|
||||
|
||||
it('rerun decides whether a completed task runs again', () => {
|
||||
const once = new Task({ description: 'd' });
|
||||
const again = new Task({ description: 'd', rerun: true });
|
||||
expect(once.needsRun()).toBe(true);
|
||||
once.markStarted();
|
||||
once.markCompleted('x');
|
||||
again.markStarted();
|
||||
again.markCompleted('x');
|
||||
expect(needsRun(once)).toBe(false); // control: no rerun -> skipped
|
||||
expect(needsRun(again)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
// ------------------------------------------------------------ scheduling
|
||||
|
||||
describe('engine/task-schedule: asyncExecution', () => {
|
||||
it('groups consecutive asyncExecution tasks into one parallel batch', () => {
|
||||
const a = { asyncExecution: true };
|
||||
const b = { asyncExecution: true };
|
||||
const c = { asyncExecution: false };
|
||||
const d = { asyncExecution: true };
|
||||
const plan = planTaskBatches([a, b, c, d]);
|
||||
expect(plan).toEqual([
|
||||
{ parallel: true, tasks: [a, b] },
|
||||
{ parallel: false, tasks: [c] },
|
||||
{ parallel: false, tasks: [d] },
|
||||
]);
|
||||
});
|
||||
|
||||
it('control: without asyncExecution every task is its own serial batch', () => {
|
||||
const tasks = [{ asyncExecution: false }, { asyncExecution: false }];
|
||||
const plan = planTaskBatches(tasks);
|
||||
expect(plan.every((b) => !b.parallel)).toBe(true);
|
||||
expect(plan).toHaveLength(2);
|
||||
});
|
||||
|
||||
it('flushes the batch when a task depends on one already queued in it', () => {
|
||||
const first = { asyncExecution: true, dependencies: [] as unknown[] };
|
||||
const second = { asyncExecution: true, dependencies: [first] };
|
||||
expect(dependsOnPending(second, [first])).toBe(true);
|
||||
expect(planTaskBatches([first, second])).toEqual([
|
||||
{ parallel: false, tasks: [first] },
|
||||
{ parallel: false, tasks: [second] },
|
||||
]);
|
||||
});
|
||||
|
||||
it('plans real Tasks through planTaskRun', () => {
|
||||
const a = new Task({ description: 'a', asyncExecution: true });
|
||||
const b = new Task({ description: 'b', asyncExecution: true });
|
||||
expect(planTaskRun([a, b])).toEqual([{ parallel: true, tasks: [a, b] }]);
|
||||
});
|
||||
});
|
||||
|
||||
// --------------------------------------------------------------- context
|
||||
|
||||
describe('engine/task-context: retainFullContext', () => {
|
||||
const previous = [
|
||||
{ name: 'one', raw: 'first result' },
|
||||
{ name: 'two', raw: 'second result' },
|
||||
];
|
||||
|
||||
it('retainFullContext includes every upstream result', () => {
|
||||
const t = new Task({ description: 'd', retainFullContext: true });
|
||||
const text = buildTaskContext(t, previous);
|
||||
expect(text).toContain('one: first result');
|
||||
expect(text).toContain('two: second result');
|
||||
});
|
||||
|
||||
it('control: without it only the most recent upstream result is included', () => {
|
||||
const t = new Task({ description: 'd' });
|
||||
const text = buildTaskContext(t, previous);
|
||||
expect(text).not.toContain('first result');
|
||||
expect(text).toContain('two: second result');
|
||||
});
|
||||
|
||||
it('folds in the task context list, respecting the same switch', () => {
|
||||
const a = new Task({ name: 'a', description: 'a' });
|
||||
a.result = { description: 'a', raw: 'A out', agent: 'x' };
|
||||
const b = new Task({ name: 'b', description: 'b' });
|
||||
b.result = { description: 'b', raw: 'B out', agent: 'x' };
|
||||
|
||||
const full = new Task({ description: 'd', retainFullContext: true, context: [a, b] });
|
||||
expect(contextOutputs(full)).toEqual([
|
||||
{ name: 'a', raw: 'A out' },
|
||||
{ name: 'b', raw: 'B out' },
|
||||
]);
|
||||
const text = buildTaskContext(full, []);
|
||||
expect(text).toContain('a: A out');
|
||||
expect(text).toContain('b: B out');
|
||||
|
||||
const recent = new Task({ description: 'd', context: [a, b] });
|
||||
const recentText = buildTaskContext(recent, []);
|
||||
expect(recentText).not.toContain('A out');
|
||||
expect(recentText).toContain('b: B out');
|
||||
});
|
||||
|
||||
it('prepends validation feedback once and then clears it', () => {
|
||||
const t = new Task({ description: 'd' });
|
||||
t.validationFeedback = { validation_response: 'not specific enough', rejected_output: 'meh' };
|
||||
const first = buildTaskContext(t, []);
|
||||
expect(first).toContain('not specific enough');
|
||||
expect(first).toContain('Rejected output: meh');
|
||||
expect(t.validationFeedback).toBeUndefined();
|
||||
expect(buildTaskContext(t, [])).toBe('');
|
||||
});
|
||||
|
||||
it('renders a bare string feedback too', () => {
|
||||
expect(renderValidationFeedback('too short')).toContain('too short');
|
||||
expect(renderValidationFeedback(undefined)).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
// --------------------------------------------------------------- routing
|
||||
|
||||
describe('engine/task-routing: when/thenTask/elseTask, condition, nextTasks, isStart', () => {
|
||||
it('routes to thenTask or elseTask on the when expression', () => {
|
||||
const t = new Task({
|
||||
description: 'd',
|
||||
when: '{{score}} > 80',
|
||||
thenTask: 'publish',
|
||||
elseTask: 'revise',
|
||||
});
|
||||
expect(resolveNextTask(t, { score: 90 })).toEqual({ kind: 'task', name: 'publish' });
|
||||
expect(resolveNextTask(t, { score: 10 })).toEqual({ kind: 'task', name: 'revise' });
|
||||
});
|
||||
|
||||
it('control: without when/then/else a task follows nextTasks', () => {
|
||||
const t = new Task({ description: 'd', nextTasks: ['summarise'] });
|
||||
expect(resolveNextTask(t, { score: 10 })).toEqual({ kind: 'task', name: 'summarise' });
|
||||
});
|
||||
|
||||
it('when-routing wins over nextTasks', () => {
|
||||
const t = new Task({ description: 'd', when: '{{ok}} == yes', thenTask: 'a', nextTasks: ['b'] });
|
||||
expect(resolveNextTask(t, { ok: 'yes' })).toEqual({ kind: 'task', name: 'a' });
|
||||
});
|
||||
|
||||
it('a when-routed branch with no target ends the path instead of falling through', () => {
|
||||
const t = new Task({ description: 'd', when: '{{ok}} == yes', thenTask: 'a' });
|
||||
expect(resolveNextTask(t, { ok: 'no' })).toEqual({ kind: 'none' });
|
||||
});
|
||||
|
||||
it('a decision task consults its condition table, and "exit" ends the workflow', () => {
|
||||
const t = new Task({
|
||||
description: 'd',
|
||||
taskType: 'decision',
|
||||
condition: { done: ['ship'], retry: ['rework'], exit: [] },
|
||||
});
|
||||
expect(decisionRoute(t, 'done')).toEqual({ kind: 'task', name: 'ship' });
|
||||
expect(decisionRoute(t, 'exit')).toEqual({ kind: 'exit' });
|
||||
expect(resolveNextTask(t, { decision: 'RETRY' })).toEqual({ kind: 'task', name: 'rework' });
|
||||
|
||||
// Control: a plain task ignores the same table and falls through.
|
||||
const plain = new Task({ description: 'd', condition: { done: ['ship'] }, nextTasks: ['other'] });
|
||||
expect(resolveNextTask(plain, { decision: 'done' })).toEqual({ kind: 'task', name: 'other' });
|
||||
});
|
||||
|
||||
it('routing is the alias for condition and drives the same table', () => {
|
||||
const t = new Task({ description: 'd', taskType: 'decision', routing: { done: ['ship'] } });
|
||||
expect(t.condition).toEqual({ done: ['ship'] });
|
||||
expect(resolveNextTask(t, { decision: 'done' })).toEqual({ kind: 'task', name: 'ship' });
|
||||
});
|
||||
|
||||
it('isStart picks the entry task, over the natural first', () => {
|
||||
const a = new Task({ name: 'a', description: 'a' });
|
||||
const b = new Task({ name: 'b', description: 'b', isStart: true });
|
||||
expect(startTaskOf([a, b])).toBe(b);
|
||||
// Control: with no flag the task nothing points at wins.
|
||||
const c = new Task({ name: 'c', description: 'c', nextTasks: ['d'] });
|
||||
const d = new Task({ name: 'd', description: 'd' });
|
||||
expect(startTaskOf([d, c])).toBe(c);
|
||||
});
|
||||
|
||||
it('nextTasks edges become previousTasks', () => {
|
||||
const a = new Task({ name: 'a', description: 'a', nextTasks: ['b'] });
|
||||
const b = new Task({ name: 'b', description: 'b', nextTasks: ['c'] });
|
||||
const c = new Task({ name: 'c', description: 'c' });
|
||||
const map = linkPreviousTasks([a, b, c]);
|
||||
expect(map.get('b')).toEqual(['a']);
|
||||
expect(map.get('c')).toEqual(['b']);
|
||||
expect(a.previousTasks).toEqual([]); // control: no incoming edge
|
||||
expect(c.previousTasks).toEqual(['b']);
|
||||
});
|
||||
});
|
||||
|
||||
// ------------------------------------------------------------ input file
|
||||
|
||||
describe('engine/task-input-file: inputFile', () => {
|
||||
it('parses quoted and escaped CSV fields', () => {
|
||||
expect(parseCsvLine('a,"b,c",d')).toEqual(['a', 'b,c', 'd']);
|
||||
expect(parseCsvLine('a\\,b,c')).toEqual(['a,b', 'c']);
|
||||
});
|
||||
|
||||
it('turns a multi-field CSV row into Python\'s Question/Answer pair', () => {
|
||||
expect(inputFileRows('q1,a1,a2\n\nq2,a3\n', '.csv')).toEqual([
|
||||
'Question: q1\nAnswer: a1,a2',
|
||||
'Question: q2\nAnswer: a3',
|
||||
]);
|
||||
});
|
||||
|
||||
it('keeps a quoted newline inside one CSV record instead of splitting it', () => {
|
||||
// A newline inside quotes is part of the field, so this is one record.
|
||||
expect(inputFileRows('q1,"line one\nline two"\nq2,a2\n', '.csv')).toEqual([
|
||||
'Question: q1\nAnswer: line one\nline two',
|
||||
'Question: q2\nAnswer: a2',
|
||||
]);
|
||||
});
|
||||
|
||||
it('decodes a doubled quote inside a quoted CSV field', () => {
|
||||
expect(inputFileRows('q1,"say ""hi"""\n', '.csv')).toEqual([
|
||||
'Question: q1\nAnswer: say "hi"',
|
||||
]);
|
||||
});
|
||||
|
||||
it('uses non-empty lines for a text file', () => {
|
||||
expect(inputFileRows('one\n\n two \n', '.txt')).toEqual(['one', 'two']);
|
||||
});
|
||||
|
||||
it('fans a task out into chained subtasks, the first flagged isStart', () => {
|
||||
const t = new Task({
|
||||
name: 'row',
|
||||
description: 'Answer',
|
||||
inputFile: 'questions.txt',
|
||||
nextTasks: ['report'],
|
||||
});
|
||||
const configs = inputFileTaskConfigs(t, { readFile: () => 'alpha\nbeta\n' });
|
||||
expect(configs).toHaveLength(2);
|
||||
expect(configs[0].name).toBe('row_1');
|
||||
expect(configs[0].description).toBe('Answer\nalpha');
|
||||
expect(configs[0].isStart).toBe(true);
|
||||
expect(configs[0].nextTasks).toEqual(['row_2']);
|
||||
expect(configs[1].isStart).toBe(false);
|
||||
// The last subtask inherits the parent's successor.
|
||||
expect(configs[1].nextTasks).toEqual(['report']);
|
||||
});
|
||||
|
||||
it('control: without inputFile nothing is expanded', () => {
|
||||
const t = new Task({ name: 'row', description: 'Answer' });
|
||||
expect(inputFileTaskConfigs(t, { readFile: () => 'alpha\n' })).toEqual([]);
|
||||
});
|
||||
|
||||
it('each input-file child inherits the parent execution settings', () => {
|
||||
const handler = () => 'done';
|
||||
const tool = () => 42;
|
||||
const t = new Task({
|
||||
name: 'row',
|
||||
description: 'Answer',
|
||||
inputFile: 'q.txt',
|
||||
handler,
|
||||
tools: [tool],
|
||||
maxRetries: 5,
|
||||
});
|
||||
const configs = inputFileTaskConfigs(t, { readFile: () => 'alpha\nbeta\n' });
|
||||
for (const config of configs) {
|
||||
expect(config.handler).toBe(handler);
|
||||
expect(config.tools).toEqual([tool]);
|
||||
expect(config.maxRetries).toBe(5);
|
||||
}
|
||||
});
|
||||
|
||||
it('decisionMode adds the done/retry/exit table Python builds', () => {
|
||||
const t = new Task({ name: 'row', description: 'Answer', inputFile: 'q.csv', nextTasks: ['report'] });
|
||||
const configs = inputFileTaskConfigs(t, { decisionMode: true, readFile: () => 'one\ntwo\n' });
|
||||
expect(configs[0].taskType).toBe('decision');
|
||||
expect(configs[0].condition).toEqual({ done: ['row_2'], retry: ['row_1'], exit: [] });
|
||||
expect(configs[1].condition).toEqual({ done: ['report'], retry: ['row_2'], exit: [] });
|
||||
});
|
||||
});
|
||||
|
||||
// ------------------------------------------------------------------ loop
|
||||
|
||||
describe('engine/task-loop: loopOver, loopVar, loopState', () => {
|
||||
it('expands one child per item, binding loopVar and recording loopState', () => {
|
||||
const t = new Task({
|
||||
name: 'greet',
|
||||
description: 'Greet {{who}}',
|
||||
loopOver: 'people',
|
||||
loopVar: 'who',
|
||||
});
|
||||
const configs = loopTaskConfigs(t, { people: ['Ada', 'Alan'] });
|
||||
expect(configs).toHaveLength(2);
|
||||
expect(configs[0].variables).toMatchObject({ who: 'Ada', _loop_index: 0 });
|
||||
expect(configs[0].loopState).toEqual({ who: 'Ada', index: 0, total: 2 });
|
||||
expect(configs[1].variables).toMatchObject({ who: 'Alan', _loop_index: 1 });
|
||||
expect(configs[1].name).toBe('greet_2');
|
||||
});
|
||||
|
||||
it('loopVar defaults to Python\'s "item"', () => {
|
||||
const t = new Task({ description: 'Process {{item}}', loopOver: 'rows' });
|
||||
const configs = loopTaskConfigs(t, { rows: [1, 2] });
|
||||
expect(configs[0].variables).toMatchObject({ item: 1 });
|
||||
expect(configs[0].loopState).toEqual({ item: 1, index: 0, total: 2 });
|
||||
});
|
||||
|
||||
it('control: no loopOver, a missing variable, or a non-list all expand to nothing', () => {
|
||||
expect(loopTaskConfigs(new Task({ description: 'd' }), { rows: [1] })).toEqual([]);
|
||||
expect(loopTaskConfigs(new Task({ description: 'd', loopOver: 'rows' }), {})).toEqual([]);
|
||||
expect(loopTaskConfigs(new Task({ description: 'd', loopOver: 'rows' }), { rows: 'nope' })).toEqual([]);
|
||||
});
|
||||
|
||||
it('each loop child inherits the parent execution settings, not just per-item fields', () => {
|
||||
const handler = () => 'done';
|
||||
const t = new Task({
|
||||
description: 'd',
|
||||
loopOver: 'rows',
|
||||
handler,
|
||||
outputJson: { type: 'object' },
|
||||
maxRetries: 7,
|
||||
caching: true,
|
||||
});
|
||||
const configs = loopTaskConfigs(t, { rows: [1, 2] });
|
||||
for (const config of configs) {
|
||||
expect(config.handler).toBe(handler);
|
||||
expect(config.outputJson).toEqual({ type: 'object' });
|
||||
expect(config.maxRetries).toBe(7);
|
||||
expect(config.caching).toBe(true);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
// -------------------------------------------------------------- messages
|
||||
|
||||
describe('engine/task-messages: images', () => {
|
||||
const fakeFs = {
|
||||
existsSync: (p: string) => p.startsWith('/local/'),
|
||||
readFileSync: () => Buffer.from('PNGDATA'),
|
||||
};
|
||||
|
||||
it('inlines a local image as a data URI and passes a URL through', () => {
|
||||
const content = buildMultimodalContent('describe', ['/local/a.png', 'https://x/y.jpg'], fakeFs);
|
||||
expect(content[0]).toEqual({ type: 'text', text: 'describe' });
|
||||
expect(content[1]).toEqual({
|
||||
type: 'image_url',
|
||||
image_url: { url: `data:image/png;base64,${Buffer.from('PNGDATA').toString('base64')}` },
|
||||
});
|
||||
expect(content[2]).toEqual({ type: 'image_url', image_url: { url: 'https://x/y.jpg' } });
|
||||
});
|
||||
|
||||
it('control: with no images the content is text only', () => {
|
||||
expect(buildMultimodalContent('describe', [], fakeFs)).toEqual([{ type: 'text', text: 'describe' }]);
|
||||
});
|
||||
|
||||
it('names a local video instead of emitting an unusable data URI', () => {
|
||||
const content = buildMultimodalContent('describe', ['/local/clip.mp4'], fakeFs);
|
||||
expect(content[1]).toEqual({ type: 'text', text: videoNote('/local/clip.mp4') });
|
||||
});
|
||||
|
||||
it('maps a local .jpg to the registered image/jpeg media type', () => {
|
||||
const content = buildMultimodalContent('describe', ['/local/a.jpg'], fakeFs);
|
||||
expect(content[1]).toEqual({
|
||||
type: 'image_url',
|
||||
image_url: { url: `data:image/jpeg;base64,${Buffer.from('PNGDATA').toString('base64')}` },
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------- output
|
||||
|
||||
describe('engine/task-output: outputPydantic, outputConfig', () => {
|
||||
it('parses the raw text when outputPydantic was requested', () => {
|
||||
const t = new Task({ description: 'd', outputPydantic: { type: 'object' } });
|
||||
const out = buildTaskOutput(t, '```json\n{"a": 1}\n```', 'writer');
|
||||
expect(out.outputFormat).toBe('Pydantic');
|
||||
expect(out.outputPydantic).toEqual({ a: 1 });
|
||||
expect(out.agent).toBe('writer');
|
||||
});
|
||||
|
||||
it('control: without a schema the output stays RAW and unparsed', () => {
|
||||
const t = new Task({ description: 'd' });
|
||||
const out = buildTaskOutput(t, '{"a": 1}');
|
||||
expect(out.outputFormat).toBe('RAW');
|
||||
expect(out.outputPydantic).toBeUndefined();
|
||||
expect(out.outputJson).toBeUndefined();
|
||||
});
|
||||
|
||||
it('records a parse failure as non-fatal and keeps the raw text', () => {
|
||||
const t = new Task({ description: 'd', outputPydantic: {} });
|
||||
const out = buildTaskOutput(t, 'not json');
|
||||
expect(out.outputFormat).toBe('RAW');
|
||||
expect(t.nonFatalErrors.join()).toContain('output parse');
|
||||
});
|
||||
|
||||
it('rejects a non-object JSON payload rather than assigning it to outputJson', () => {
|
||||
const t = new Task({ description: 'd', outputJson: { type: 'object' } });
|
||||
const out = buildTaskOutput(t, '[1, 2, 3]');
|
||||
// A top-level array is valid JSON but not a Record; it must not be
|
||||
// labelled JSON output, and it must not be assigned.
|
||||
expect(out.outputFormat).toBe('RAW');
|
||||
expect(out.outputJson).toBeUndefined();
|
||||
expect(t.nonFatalErrors.join()).toContain('expected a JSON object');
|
||||
});
|
||||
|
||||
it('cleanJsonOutput strips a markdown fence', () => {
|
||||
expect(cleanJsonOutput('```json\n{"a":1}\n```')).toBe('{"a":1}');
|
||||
expect(cleanJsonOutput('{"a":1}')).toBe('{"a":1}');
|
||||
});
|
||||
|
||||
it('resolveOutputConfig reads a string as a file and an object field by field', () => {
|
||||
expect(resolveOutputConfig('out.md')).toEqual({ file: 'out.md' });
|
||||
expect(resolveOutputConfig({ file: 'f', json_model: { a: 1 }, variable: 'v' }))
|
||||
.toEqual({ file: 'f', json: { a: 1 }, variable: 'v' });
|
||||
expect(resolveOutputConfig({})).toBeUndefined();
|
||||
expect(resolveOutputConfig(undefined)).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
// --------------------------------------------------------------- handler
|
||||
|
||||
describe('engine/task-handler: handler', () => {
|
||||
it('runs the handler instead of an agent and stringifies a plain return', async () => {
|
||||
const t = new Task({ name: 'fn', handler: (ctx: unknown) => `saw:${(ctx as { currentStep?: string }).currentStep}` });
|
||||
const result = await runTaskHandler(t, { variables: {}, currentStep: 'fn' });
|
||||
expect(result).toEqual({ success: true, output: 'saw:fn', stop: false });
|
||||
});
|
||||
|
||||
it('honours a StepResult return: output, variables and stop_workflow', async () => {
|
||||
const t = new Task({ name: 'fn', handler: () => ({ output: 'ok', variables: { n: 3 }, stopWorkflow: true }) });
|
||||
const result = await runTaskHandler(t, { variables: {} });
|
||||
expect(result).toEqual({ success: true, output: 'ok', variables: { n: 3 }, stop: true });
|
||||
});
|
||||
|
||||
it('awaits an async handler', async () => {
|
||||
const t = new Task({ name: 'fn', handler: async () => 'later' });
|
||||
expect((await runTaskHandler(t, { variables: {} }))?.output).toBe('later');
|
||||
});
|
||||
|
||||
it('turns a throwing handler into a failed step, not a failed run', async () => {
|
||||
const t = new Task({ name: 'fn', handler: () => { throw new Error('boom'); } });
|
||||
expect(await runTaskHandler(t, { variables: {} })).toEqual({
|
||||
success: false, output: null, stop: false, error: 'boom',
|
||||
});
|
||||
});
|
||||
|
||||
it('control: a task without a handler returns undefined so the agent path runs', async () => {
|
||||
const t = new Task({ description: 'd' });
|
||||
expect(await runTaskHandler(t, { variables: {} })).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
// ----------------------------------------------------------------- agent
|
||||
|
||||
describe('engine/task-agent: agentConfig', () => {
|
||||
it('builds an agent from agentConfig and assigns it to the task', () => {
|
||||
const built: Array<Record<string, unknown>> = [];
|
||||
const t = new Task({
|
||||
name: 'research',
|
||||
description: 'd',
|
||||
agentConfig: { role: 'Researcher', goal: 'find things', verbose: true },
|
||||
tools: ['search'],
|
||||
});
|
||||
const agent = resolveTaskAgent(t, {
|
||||
defaultLlm: 'gpt-test',
|
||||
createAgent: (opts) => { built.push(opts); return { id: 'built', ...opts }; },
|
||||
});
|
||||
expect(built[0]).toEqual({
|
||||
role: 'Researcher', goal: 'find things', name: 'researchAgent', llm: 'gpt-test', tools: ['search'],
|
||||
});
|
||||
expect(t.agent).toBe(agent);
|
||||
});
|
||||
|
||||
it('control: without agentConfig the caller default is used and nothing is built', () => {
|
||||
const t = new Task({ description: 'd' });
|
||||
const fallback = { name: 'default' };
|
||||
const createAgent = jest.fn();
|
||||
expect(resolveTaskAgent(t, { defaultAgent: fallback, createAgent })).toBe(fallback);
|
||||
expect(createAgent).not.toHaveBeenCalled();
|
||||
expect(agentOptionsFor(t)).toBeUndefined();
|
||||
});
|
||||
|
||||
it('the task\'s own agent always wins over agentConfig', () => {
|
||||
const own = { name: 'own' };
|
||||
const t = new Task({ description: 'd', agent: own, agentConfig: { role: 'x' } });
|
||||
const createAgent = jest.fn();
|
||||
expect(resolveTaskAgent(t, { createAgent })).toBe(own);
|
||||
expect(createAgent).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('falls back to the default agent when construction throws', () => {
|
||||
const t = new Task({ description: 'd', agentConfig: { role: 'x' } });
|
||||
const fallback = { name: 'default' };
|
||||
expect(resolveTaskAgent(t, { defaultAgent: fallback, createAgent: () => { throw new Error('no'); } }))
|
||||
.toBe(fallback);
|
||||
});
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------- memory
|
||||
|
||||
describe('engine/task-memory: memory, config.memory_config, failOnMemoryError', () => {
|
||||
function fakeStore() {
|
||||
return {
|
||||
added: [] as string[],
|
||||
add(content: string) { this.added.push(content); return Promise.resolve(content); },
|
||||
search: () => Promise.resolve([{ entry: { content: 'remembered' }, score: 1 }]),
|
||||
};
|
||||
}
|
||||
|
||||
it('stores the output in the task memory, and recalls it into the next prompt', async () => {
|
||||
const store = fakeStore();
|
||||
const t = new Task({ description: 'd', memory: store });
|
||||
expect(await storeTaskOutput(t, 'the answer', 'writer')).toBe(true);
|
||||
expect(store.added).toEqual(['the answer']);
|
||||
expect(await buildMemoryContext(t, 'question')).toBe('remembered');
|
||||
});
|
||||
|
||||
it('control: a task with no memory stores nothing and recalls nothing', async () => {
|
||||
const t = new Task({ description: 'd' });
|
||||
expect(await storeTaskOutput(t, 'the answer')).toBe(false);
|
||||
expect(await buildMemoryContext(t, 'question')).toBe('');
|
||||
});
|
||||
|
||||
it('config.memory_config builds the store lazily', () => {
|
||||
const t = new Task({ description: 'd', config: { memory_config: { provider: 'local' } } });
|
||||
expect(memoryConfigOf(t)).toEqual({ provider: 'local' });
|
||||
const store = fakeStore();
|
||||
expect(t.initializeMemory(() => store)).toBe(store);
|
||||
expect(t.memory).toBe(store);
|
||||
|
||||
// Control: no memory_config, no store.
|
||||
const plain = new Task({ description: 'd', config: { verbose: 5 } });
|
||||
expect(memoryConfigOf(plain)).toBeUndefined();
|
||||
expect(plain.initializeMemory(() => store)).toBeUndefined();
|
||||
});
|
||||
|
||||
it('failOnMemoryError turns a swallowed store failure into a thrown one', async () => {
|
||||
const bad = { add: () => { throw new Error('disk full'); }, search: () => Promise.resolve([]) };
|
||||
|
||||
const lenient = new Task({ description: 'd', memory: bad });
|
||||
expect(await storeTaskOutput(lenient, 'x')).toBe(false);
|
||||
expect(lenient.nonFatalErrors.join()).toContain('disk full');
|
||||
|
||||
const strict = new Task({ description: 'd', memory: bad, failOnMemoryError: true });
|
||||
await expect(storeTaskOutput(strict, 'x')).rejects.toThrow('disk full');
|
||||
});
|
||||
});
|
||||
|
||||
// -------------------------------------------------------------- features
|
||||
|
||||
describe('engine/task-features: hooks, caching, knowledge', () => {
|
||||
it('resolves hooks from camelCase, snake_case and the on_step_* spelling', () => {
|
||||
const fn = (): void => undefined;
|
||||
expect(resolveTaskHooks({ onTaskStart: fn })?.onTaskStart).toBe(fn);
|
||||
expect(resolveTaskHooks({ on_task_complete: fn })?.onTaskComplete).toBe(fn);
|
||||
expect(resolveTaskHooks({ on_step_start: fn })?.onTaskStart).toBe(fn);
|
||||
// Control: nothing callable, nothing resolved.
|
||||
expect(resolveTaskHooks(undefined)).toBeUndefined();
|
||||
expect(resolveTaskHooks({ notAHook: 1 })).toBeUndefined();
|
||||
expect(resolveTaskHooks([])).toBeUndefined();
|
||||
});
|
||||
|
||||
it('caching resolves to a cache only when enabled', () => {
|
||||
expect(resolveTaskCache(true)).toBeDefined();
|
||||
expect(resolveTaskCache('enabled')).toBeDefined();
|
||||
expect(resolveTaskCache({ enabled: true, ttl: 60 })).toBeDefined();
|
||||
// Control.
|
||||
expect(resolveTaskCache(undefined)).toBeUndefined();
|
||||
expect(resolveTaskCache(false)).toBeUndefined();
|
||||
expect(resolveTaskCache('disabled')).toBeUndefined();
|
||||
});
|
||||
|
||||
it('the resolved cache round-trips a prompt answer', () => {
|
||||
const cache = resolveTaskCache(true)!;
|
||||
cache.set(cacheKey('writer', 'hello'), 'cached answer');
|
||||
expect(cache.get(cacheKey('writer', 'hello'))).toBe('cached answer');
|
||||
expect(cache.get(cacheKey('other', 'hello'))).toBeUndefined();
|
||||
});
|
||||
|
||||
it('knowledge contributes literal text, a document list, or search hits', async () => {
|
||||
const literal = new Task({ description: 'd', knowledge: 'the sky is blue' });
|
||||
expect(await buildKnowledgeContext(literal, 'sky')).toBe('the sky is blue');
|
||||
|
||||
const list = new Task({ description: 'd', knowledge: ['one', 'two'] });
|
||||
expect(await buildKnowledgeContext(list, 'x')).toBe('one\ntwo');
|
||||
|
||||
const searched = new Task({
|
||||
description: 'd',
|
||||
knowledge: { search: async () => [{ document: { content: 'found it' }, score: 1 }] },
|
||||
});
|
||||
expect(await buildKnowledgeContext(searched, 'x')).toBe('found it');
|
||||
|
||||
// The canonical `Knowledge.search` returns `{ results: SearchResultItem[] }`
|
||||
// where each item carries `text`, not `{ document: { content } }`.
|
||||
const canonical = new Task({
|
||||
description: 'd',
|
||||
knowledge: {
|
||||
search: async () => ({
|
||||
results: [{ id: '1', text: 'canonical hit', score: 0.9, metadata: {} }],
|
||||
metadata: {},
|
||||
query: 'x',
|
||||
}),
|
||||
},
|
||||
});
|
||||
expect(await buildKnowledgeContext(canonical, 'x')).toBe('canonical hit');
|
||||
});
|
||||
|
||||
it('control: no knowledge, no block; a failing search is non-fatal', async () => {
|
||||
expect(await buildKnowledgeContext(new Task({ description: 'd' }), 'x')).toBe('');
|
||||
const broken = new Task({
|
||||
description: 'd',
|
||||
knowledge: { search: async () => { throw new Error('index down'); } },
|
||||
});
|
||||
expect(await buildKnowledgeContext(broken, 'x')).toBe('');
|
||||
expect(broken.nonFatalErrors.join()).toContain('index down');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,346 @@
|
||||
/**
|
||||
* Behaviour parity for `Task.__init__` options: proof that each option changes
|
||||
* what the code does, through the public methods a team runner calls.
|
||||
*
|
||||
* Python: src/praisonai-agents/praisonaiagents/task/task.py, plus the code in
|
||||
* agents/agents.py, process/process.py and workflows/workflows.py that consults
|
||||
* each field while running a task.
|
||||
*/
|
||||
process.env.PRAISONAI_PARITY_SILENT = '1';
|
||||
|
||||
import { Task } from '../../../src/agent/types';
|
||||
import type { TaskCallbackMetadata, TaskOutput } from '../../../src/agent/types';
|
||||
import { planTaskRun } from '../../../src/agent/engine/task-schedule';
|
||||
import { linkPreviousTasks, startTaskOf } from '../../../src/agent/engine/task-routing';
|
||||
|
||||
function makeOutput(raw = 'r'): TaskOutput {
|
||||
return { description: 'd', raw, agent: 'a' };
|
||||
}
|
||||
|
||||
describe('Task behaviour: retries and failure policy', () => {
|
||||
it('retryDelay backs off exponentially; the default never waits', async () => {
|
||||
const slow = new Task({ description: 'd', retryDelay: 3 });
|
||||
const fast = new Task({ description: 'd' });
|
||||
expect(slow.retryDelayFor(0)).toBe(3);
|
||||
expect(slow.retryDelayFor(2)).toBe(12);
|
||||
expect(fast.retryDelayFor(2)).toBe(0);
|
||||
|
||||
const waits: number[] = [];
|
||||
const timer = (fn: () => void, ms: number): unknown => { waits.push(ms); fn(); return 0; };
|
||||
expect(await slow.waitBeforeRetry(1, timer)).toBe(6);
|
||||
expect(await fast.waitBeforeRetry(1, timer)).toBe(0);
|
||||
expect(waits).toEqual([6000]); // the control scheduled nothing
|
||||
});
|
||||
|
||||
it('skipOnFailure keeps the task runnable after an upstream failure', () => {
|
||||
expect(new Task({ description: 'd', skipOnFailure: true }).continuesOnDependencyFailure()).toBe(true);
|
||||
expect(new Task({ description: 'd' }).continuesOnDependencyFailure()).toBe(false);
|
||||
});
|
||||
|
||||
it('rerun re-runs a completed task; without it the task is skipped', () => {
|
||||
const complete = (t: Task): Task => { t.markStarted(); t.markCompleted('x'); return t; };
|
||||
expect(complete(new Task({ description: 'd', rerun: true })).needsRun()).toBe(true);
|
||||
expect(complete(new Task({ description: 'd' })).needsRun()).toBe(false);
|
||||
});
|
||||
|
||||
it('execution config fills asyncExecution, rerun and the retry policy', () => {
|
||||
const t = new Task({ description: 'd', execution: { asyncExec: true, rerun: true, max_retries: 7 } });
|
||||
expect(t.asyncExecution).toBe(true);
|
||||
expect(t.rerun).toBe(true);
|
||||
expect(t.maxRetries).toBe(7);
|
||||
expect(planTaskRun([t])).toEqual([{ parallel: false, tasks: [t] }]);
|
||||
|
||||
// Control: no execution config, Python's defaults, and a direct param wins.
|
||||
const control = new Task({ description: 'd' });
|
||||
expect(control.asyncExecution).toBe(false);
|
||||
expect(control.maxRetries).toBe(3);
|
||||
const direct = new Task({ description: 'd', maxRetries: 1, execution: { max_retries: 7 } });
|
||||
expect(direct.maxRetries).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('Task behaviour: context and routing', () => {
|
||||
it('retainFullContext changes how much upstream output reaches the prompt', () => {
|
||||
const previous = [{ name: 'a', raw: 'A' }, { name: 'b', raw: 'B' }];
|
||||
expect(new Task({ description: 'd', retainFullContext: true }).buildContext(previous))
|
||||
.toContain('a: A');
|
||||
expect(new Task({ description: 'd' }).buildContext(previous)).not.toContain('a: A');
|
||||
});
|
||||
|
||||
it('when/thenTask/elseTask route the workflow', () => {
|
||||
const t = new Task({ description: 'd', when: '{{score}} >= 80', thenTask: 'ship', elseTask: 'revise' });
|
||||
expect(t.evaluateWhen({ score: 95 })).toBe(true);
|
||||
expect(t.nextTaskFor({ score: 95 })).toBe('ship');
|
||||
expect(t.nextTaskFor({ score: 12 })).toBe('revise');
|
||||
|
||||
// Control: no `when` always passes, and routes only via nextTasks.
|
||||
const control = new Task({ description: 'd', nextTasks: ['always'] });
|
||||
expect(control.evaluateWhen({ score: 12 })).toBe(true);
|
||||
expect(control.nextTaskFor({ score: 12 })).toBe('always');
|
||||
});
|
||||
|
||||
it('condition/routing drive a decision task, and exit ends the run', () => {
|
||||
const t = new Task({ description: 'd', taskType: 'decision', routing: { done: ['ship'], exit: [] } });
|
||||
expect(t.nextTaskFor({ decision: 'done' })).toBe('ship');
|
||||
expect(t.routeAfter({ decision: 'exit' })).toEqual({ kind: 'exit' });
|
||||
|
||||
// Control: no table, no route.
|
||||
const control = new Task({ description: 'd', taskType: 'decision' });
|
||||
expect(control.routeAfter({ decision: 'done' })).toEqual({ kind: 'none' });
|
||||
});
|
||||
|
||||
it('isStart and nextTasks lay out the workflow graph', () => {
|
||||
const first = new Task({ name: 'first', description: 'a', nextTasks: ['second'] });
|
||||
const second = new Task({ name: 'second', description: 'b', isStart: true });
|
||||
expect(startTaskOf([first, second])).toBe(second); // isStart wins
|
||||
linkPreviousTasks([first, second]);
|
||||
expect(second.previousTasks).toEqual(['first']);
|
||||
expect(first.previousTasks).toEqual([]); // control
|
||||
});
|
||||
});
|
||||
|
||||
describe('Task behaviour: fan-out', () => {
|
||||
it('loopOver/loopVar expand into one task per item, each carrying loopState', () => {
|
||||
const t = new Task({ name: 'greet', description: 'Greet {{who}}', loopOver: 'people', loopVar: 'who' });
|
||||
const children = t.expandLoop({ people: ['Ada', 'Alan'] });
|
||||
expect(children).toHaveLength(2);
|
||||
expect(children[0].renderDescription()).toBe('Greet Ada');
|
||||
expect(children[1].renderDescription()).toBe('Greet Alan');
|
||||
expect(children[1].loopState).toEqual({ who: 'Alan', index: 1, total: 2 });
|
||||
|
||||
// Control: no loopOver, no fan-out, and the placeholder stays.
|
||||
const control = new Task({ description: 'Greet {{who}}' });
|
||||
expect(control.expandLoop({ people: ['Ada'] })).toEqual([]);
|
||||
expect(control.renderDescription()).toBe('Greet {{who}}');
|
||||
});
|
||||
|
||||
it('inputFile expands into chained subtasks', () => {
|
||||
const t = new Task({ name: 'ask', description: 'Answer', inputFile: 'q.txt', nextTasks: ['report'] });
|
||||
const children = t.expandFromInputFile({ readFile: () => 'alpha\nbeta\n' });
|
||||
expect(children.map((c) => c.name)).toEqual(['ask_1', 'ask_2']);
|
||||
expect(children[0].description).toBe('Answer\nalpha');
|
||||
expect(children[0].isStart).toBe(true);
|
||||
expect(children[1].nextTasks).toEqual(['report']);
|
||||
|
||||
// Control: no inputFile, no subtasks.
|
||||
expect(new Task({ name: 'ask', description: 'Answer' })
|
||||
.expandFromInputFile({ readFile: () => 'alpha\n' })).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('Task behaviour: execution surface', () => {
|
||||
it('handler runs in place of an agent', async () => {
|
||||
const t = new Task({ name: 'fn', handler: (ctx: unknown) => `hi ${(ctx as { variables: Record<string, unknown> }).variables.who}` });
|
||||
expect(await t.runHandler({ variables: { who: 'Ada' } })).toEqual({
|
||||
success: true, output: 'hi Ada', stop: false,
|
||||
});
|
||||
// Control: no handler -> undefined, so the caller uses the agent.
|
||||
expect(await new Task({ description: 'd' }).runHandler({ variables: {} })).toBeUndefined();
|
||||
});
|
||||
|
||||
it('agentConfig builds the executing agent when the task names none', () => {
|
||||
const t = new Task({ name: 'research', description: 'd', agentConfig: { role: 'Researcher' } });
|
||||
const agent = t.resolveAgent({ defaultLlm: 'm', createAgent: (o) => ({ built: o }) });
|
||||
expect(agent).toEqual({ built: { role: 'Researcher', name: 'researchAgent', llm: 'm' } });
|
||||
expect(t.agent).toBe(agent);
|
||||
|
||||
// Control: no agentConfig -> the caller's default, nothing constructed.
|
||||
const control = new Task({ description: 'd' });
|
||||
const createAgent = jest.fn();
|
||||
expect(control.resolveAgent({ defaultAgent: 'fallback', createAgent })).toBe('fallback');
|
||||
expect(createAgent).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('images turn the prompt into multimodal content', () => {
|
||||
const fakeFs = { existsSync: () => false, readFileSync: () => Buffer.from('') };
|
||||
const withImages = new Task({ description: 'd', images: ['https://x/y.png'] });
|
||||
expect(withImages.buildMessageContent('look', fakeFs)).toEqual([
|
||||
{ type: 'text', text: 'look' },
|
||||
{ type: 'image_url', image_url: { url: 'https://x/y.png' } },
|
||||
]);
|
||||
// Control: no images -> the prompt string is passed through unchanged.
|
||||
expect(new Task({ description: 'd' }).buildMessageContent('look', fakeFs)).toBe('look');
|
||||
});
|
||||
|
||||
it('outputPydantic parses the agent text into structured output', () => {
|
||||
const structured = new Task({ description: 'd', outputPydantic: { type: 'object' } });
|
||||
const out = structured.buildOutput('{"ok": true}', 'writer');
|
||||
expect(out.outputFormat).toBe('Pydantic');
|
||||
expect(out.outputPydantic).toEqual({ ok: true });
|
||||
|
||||
// Control: without a schema the same text stays RAW.
|
||||
const control = new Task({ description: 'd' });
|
||||
expect(control.buildOutput('{"ok": true}').outputFormat).toBe('RAW');
|
||||
expect(control.buildOutput('{"ok": true}').outputPydantic).toBeUndefined();
|
||||
});
|
||||
|
||||
it('outputConfig fills the output fields, and an explicit output wins over it', () => {
|
||||
const fromString = new Task({ description: 'd', outputConfig: 'report.md' });
|
||||
expect(fromString.outputFile).toBe('report.md');
|
||||
|
||||
const fromObject = new Task({
|
||||
description: 'd',
|
||||
outputConfig: { file: 'a.md', pydantic_model: { type: 'object' }, variable: 'answer' },
|
||||
});
|
||||
expect(fromObject.outputFile).toBe('a.md');
|
||||
expect(fromObject.outputPydantic).toEqual({ type: 'object' });
|
||||
expect(fromObject.outputVariable).toBe('answer');
|
||||
|
||||
const overridden = new Task({ description: 'd', outputConfig: { file: 'a.md' }, output: 'b.md' });
|
||||
expect(overridden.outputFile).toBe('b.md');
|
||||
|
||||
// An explicit individual param also wins over the config object.
|
||||
const explicit = new Task({ description: 'd', outputConfig: { file: 'a.md' }, outputFile: 'c.md' });
|
||||
expect(explicit.outputFile).toBe('c.md');
|
||||
|
||||
// Control: no outputConfig, no output file.
|
||||
expect(new Task({ description: 'd' }).outputFile).toBeUndefined();
|
||||
});
|
||||
|
||||
it('asyncExecution batches consecutive tasks for a parallel fan-out', () => {
|
||||
const a = new Task({ description: 'a', asyncExecution: true });
|
||||
const b = new Task({ description: 'b', asyncExecution: true });
|
||||
const c = new Task({ description: 'c' });
|
||||
expect(planTaskRun([a, b, c])).toEqual([
|
||||
{ parallel: true, tasks: [a, b] },
|
||||
{ parallel: false, tasks: [c] },
|
||||
]);
|
||||
// Control: three plain tasks stay serial.
|
||||
const plain = [new Task({ description: 'a' }), new Task({ description: 'b' })];
|
||||
expect(planTaskRun(plain).every((batch) => !batch.parallel)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('Task behaviour: memory, knowledge, hooks and caching', () => {
|
||||
function store() {
|
||||
return {
|
||||
added: [] as string[],
|
||||
add(content: string) { this.added.push(content); return Promise.resolve(content); },
|
||||
search: () => Promise.resolve([{ entry: { content: 'recalled' }, score: 1 }]),
|
||||
};
|
||||
}
|
||||
|
||||
it('memory stores the output and recalls it; a task without memory does neither', async () => {
|
||||
const mem = store();
|
||||
const t = new Task({ description: 'd', memory: mem });
|
||||
expect(await t.storeInMemory('answer', 'writer')).toBe(true);
|
||||
expect(mem.added).toEqual(['answer']);
|
||||
expect(await t.memoryContext('q')).toBe('recalled');
|
||||
|
||||
const control = new Task({ description: 'd' });
|
||||
expect(await control.storeInMemory('answer')).toBe(false);
|
||||
expect(await control.memoryContext('q')).toBe('');
|
||||
});
|
||||
|
||||
it('config.memory_config creates the store on first use', () => {
|
||||
const mem = store();
|
||||
const t = new Task({ description: 'd', config: { memory_config: { provider: 'local' } } });
|
||||
expect(t.memory).toBeUndefined();
|
||||
expect(t.initializeMemory(() => mem)).toBe(mem);
|
||||
expect(t.verboseLevel).toBe(0);
|
||||
|
||||
const verbose = new Task({ description: 'd', config: { verbose: 5 } });
|
||||
expect(verbose.verboseLevel).toBe(5);
|
||||
// Control: no memory_config -> nothing built.
|
||||
expect(verbose.initializeMemory(() => mem)).toBeUndefined();
|
||||
});
|
||||
|
||||
it('failOnMemoryError decides whether a memory failure stops the task', async () => {
|
||||
const broken = { add: () => { throw new Error('disk full'); }, search: () => Promise.resolve([]) };
|
||||
const lenient = new Task({ description: 'd', memory: broken });
|
||||
expect(await lenient.storeInMemory('x')).toBe(false);
|
||||
expect(lenient.nonFatalErrors.join()).toContain('disk full');
|
||||
|
||||
const strict = new Task({ description: 'd', memory: broken, failOnMemoryError: true });
|
||||
await expect(strict.storeInMemory('x')).rejects.toThrow('disk full');
|
||||
});
|
||||
|
||||
it('knowledge contributes a block to the prompt', async () => {
|
||||
expect(await new Task({ description: 'd', knowledge: 'sky is blue' }).knowledgeContext('q'))
|
||||
.toBe('sky is blue');
|
||||
expect(await new Task({ description: 'd' }).knowledgeContext('q')).toBe('');
|
||||
});
|
||||
|
||||
it('hooks fire around the task', async () => {
|
||||
const seen: string[] = [];
|
||||
const t = new Task({
|
||||
name: 'hooked',
|
||||
description: 'd',
|
||||
hooks: {
|
||||
onTaskStart: (task: Task, index: number) => { seen.push(`start:${task.name}:${index}`); },
|
||||
on_task_complete: (task: Task, out: TaskOutput) => { seen.push(`done:${out.raw}`); },
|
||||
},
|
||||
});
|
||||
expect(t.resolvedHooks).toBeDefined();
|
||||
await t.notifyStart(2);
|
||||
await t.notifyComplete(makeOutput('final'));
|
||||
expect(seen).toEqual(['start:hooked:2', 'done:final']);
|
||||
|
||||
// Control: no hooks, nothing fires and nothing breaks.
|
||||
const control = new Task({ description: 'd' });
|
||||
expect(control.resolvedHooks).toBeUndefined();
|
||||
await control.notifyStart();
|
||||
await control.notifyComplete(makeOutput());
|
||||
expect(seen).toHaveLength(2);
|
||||
});
|
||||
|
||||
it('a throwing hook is non-fatal unless failOnCallbackError is set', async () => {
|
||||
const t = new Task({ description: 'd', hooks: { onTaskComplete: () => { throw new Error('hook boom'); } } });
|
||||
const out = await t.notifyComplete(makeOutput());
|
||||
expect(out.nonFatalErrors?.join()).toContain('hook boom');
|
||||
|
||||
const strict = new Task({
|
||||
description: 'd',
|
||||
failOnCallbackError: true,
|
||||
hooks: { onTaskComplete: () => { throw new Error('hook boom'); } },
|
||||
});
|
||||
await expect(strict.notifyComplete(makeOutput())).rejects.toThrow('hook boom');
|
||||
});
|
||||
|
||||
it('caching gives the task a response cache; without it there is none', () => {
|
||||
const cached = new Task({ description: 'd', caching: true });
|
||||
expect(cached.cache).toBeDefined();
|
||||
cached.cache!.set('k', 'v');
|
||||
expect(cached.cache!.get('k')).toBe('v');
|
||||
expect(new Task({ description: 'd' }).cache).toBeUndefined();
|
||||
expect(new Task({ description: 'd', caching: 'disabled' }).cache).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('Task behaviour: callback metadata', () => {
|
||||
it('a two-parameter callback receives loopState, inputFile and asyncExecution', async () => {
|
||||
let metadata: TaskCallbackMetadata | undefined;
|
||||
const t = new Task({
|
||||
name: 'row_2',
|
||||
description: 'd',
|
||||
inputFile: 'rows.csv',
|
||||
loopState: { item: 'beta', index: 1 },
|
||||
asyncExecution: true,
|
||||
taskType: 'loop',
|
||||
onTaskComplete: (_out: TaskOutput, meta?: TaskCallbackMetadata) => { metadata = meta; },
|
||||
});
|
||||
await t.notifyComplete(makeOutput());
|
||||
expect(metadata).toMatchObject({
|
||||
taskName: 'row_2',
|
||||
inputFile: 'rows.csv',
|
||||
loopState: { item: 'beta', index: 1 },
|
||||
asyncExecution: true,
|
||||
taskType: 'loop',
|
||||
retryCount: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it('control: metadata is always passed; a one-parameter callback ignores it', async () => {
|
||||
const args: unknown[][] = [];
|
||||
const t = new Task({
|
||||
description: 'd',
|
||||
onTaskComplete: (...received: unknown[]) => { args.push(received); },
|
||||
});
|
||||
// Metadata is handed to every callback as a second argument; a callback
|
||||
// that declares fewer parameters simply ignores it, as in JavaScript.
|
||||
await t.notifyComplete(makeOutput('only'));
|
||||
expect(args).toHaveLength(1);
|
||||
expect(args[0][0]).toEqual({ description: 'd', raw: 'only', agent: 'a' });
|
||||
expect(args[0][1]).toMatchObject({ taskDescription: 'd', retryCount: 0 });
|
||||
});
|
||||
});
|
||||
@@ -11,7 +11,7 @@ import * as os from 'os';
|
||||
import * as path from 'path';
|
||||
import { Task, TASK_STATUS } from '../../../src/agent/types';
|
||||
import type { TaskConfig, TaskOutput } from '../../../src/agent/types';
|
||||
import { resetParityNotices, unhonouredOptions } from '../../../src/utils/parity-notice';
|
||||
import { resetParityNotices, unhonouredFor, unhonouredOptions } from '../../../src/utils/parity-notice';
|
||||
|
||||
function makeOutput(overrides: Partial<TaskOutput> = {}): TaskOutput {
|
||||
return { description: 'd', raw: 'r', agent: 'a', ...overrides };
|
||||
@@ -505,15 +505,16 @@ describe('Task parity: engine-level options are not silently ignored', () => {
|
||||
taskType: 'loop',
|
||||
guardrails: 'llm judged',
|
||||
});
|
||||
expect(unhonouredOptions()).toEqual([
|
||||
'Task.agentConfig', 'Task.asyncExecution', 'Task.autonomy', 'Task.caching', 'Task.condition',
|
||||
'Task.config', 'Task.elseTask', 'Task.execution', 'Task.failOnMemoryError', 'Task.guardrails',
|
||||
'Task.handler', 'Task.hooks', 'Task.images', 'Task.inputFile', 'Task.isStart', 'Task.knowledge',
|
||||
'Task.loopOver', 'Task.loopState', 'Task.loopVar', 'Task.memory', 'Task.nextTasks',
|
||||
'Task.outputConfig', 'Task.outputPydantic', 'Task.planning', 'Task.reflection', 'Task.rerun',
|
||||
'Task.retainFullContext', 'Task.retryDelay', 'Task.routing', 'Task.skipOnFailure',
|
||||
'Task.taskType', 'Task.thenTask', 'Task.web', 'Task.when',
|
||||
]);
|
||||
// Derived from the ledger rather than hard-coded: the assertion is that the
|
||||
// constructor reports EVERY option still listed as unhonoured (plus the two
|
||||
// locally-detected ones), so closing an option shrinks both together and
|
||||
// this test cannot drift from utils/parity-notice.ts.
|
||||
const expected = [
|
||||
...unhonouredFor('Task.__init__').map((option) => `Task.${option}`),
|
||||
'Task.taskType',
|
||||
'Task.guardrails',
|
||||
].sort();
|
||||
expect(unhonouredOptions()).toEqual(expected);
|
||||
});
|
||||
|
||||
it('does not register locally-honoured options', () => {
|
||||
|
||||
@@ -152,7 +152,9 @@ describe('Workflow', () => {
|
||||
});
|
||||
|
||||
const { results } = await workflow.run();
|
||||
expect(results[0].duration).toBeGreaterThanOrEqual(50);
|
||||
// Allow a small tolerance below the 50ms sleep: timer scheduling and
|
||||
// Date rounding can report a duration a millisecond or two short.
|
||||
expect(results[0].duration).toBeGreaterThanOrEqual(45);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
Reference in new issue
Block a user