Process observation and node reports

The App subscribes to Core execution facts; nodes report their own usedModel and use an instance sub-logger to record intent first and then results, while the Loop owns identity and logging.

execution.observe.event is side-channel observation: observer failures do not change task results and do not create delivery backpressure. The model body sink.output is a different port; Core awaits it in order and propagates delivery failures. Do not treat a side-channel log observer as a reliable terminal output queue.

A Workflow node calls context.report({usedModel, payload}) during execution, or provides report(update, context) to describe completion details. usedModel indicates the model usage reported by the node itself; when not reported it is unknown, and child nodes are not aggregated. Sub-logs must be created as instances via context.createSubFlowLogger(); write returns an ID on success or -1 on failure, fix returns true on success or false on failure, and readAll reads the full immutable ledger; hand-written arrays or plain objects are not accepted. write records into the existing node.progress immediately, and a short circuit still keeps the plan; fix appends a revision without rewriting historical events. Nodes do not create run identity and do not repeatedly pass diagnostics; Core automatically records node entry, transitions, errors, and model and tool facts. The App decides which statuses to show users; not every Provider has or allows exposure of the model's internal reasoning text.

write requires name and kind, status defaults to planned; the ID, initial order, createdAtMs, and revision are generated by the instance. running means execution actually started, while success, error, cancelled, and skipped are terminal states. fix cannot reopen terminal states or overwrite established facts, but it can fill in missing results while the current execution is still active. payload accepts JSON data, and parent steps and successor IDs must belong to the same node visit ledger; readAll is a persistent readonly query, and records are not submitted through report.

After node completion or cancellation, write returns -1 and fix returns false; retained instances remain readable. fork(subnodeId) creates an independent ScopedSubFlowLogger delegate without storage of its own. It binds writes to the group and only fixes that group's original writes; it cannot fork or be passed to invoke. readAll exposes the full shared ledger. subnodeId and business type are caller-defined; operation separately identifies write or fix.

The simple agent.run(step) and tool execution contexts also provide report and createSubFlowLogger. The tool's debugSteps(result, context) records directly through its context logger without returning or submitting it; the Workflow completion callback report(update, context) returns the report object. Both completion callbacks are evaluated only when content collection is permitted, and metadata marks should be submitted during execution via context.report. subSteps in logs is a normalized array, and payload.subStep is the progress metadata of write/fix; consumers without content authorization receive only approved metadata.

createDiagnostics configures optional log consumption of the same fact source; it is not another persistent journal. Body text remains governed by the content recording policy. The run/span/invocation associations in events are Core execution identities, not App message IDs; the App correlates its own message records.

Progressive examples for the same business case

Observe the nodes and model events a task passes through

Beginner · Node.js

TypeScript

import { AgentCoreEngine, defineWorkflow, route } from '@jangts/agent-core/engine';

type State = { name: string; answer?: string };
const workflow = defineWorkflow<State, string>({
  id: 'greeting',
  nodes: [{
    id: 'greet', description: '根据本次姓名生成问候', startsRound: true,
    run: state => ({ answer: '你好，' + state.name }),
    routes: [route.deliver('answer')],
  }],
});
const agent = new AgentCoreEngine({ workflow, execution: {
  observe: { event(event) { console.log(event.type, event.eventId); } },
} });
console.log(await agent.run({ name: 'Ivan' }));

JavaScript

import { AgentCoreEngine, defineWorkflow, route } from '@jangts/agent-core/engine';
const workflow = defineWorkflow({
    id: 'greeting',
    nodes: [{
            id: 'greet', description: '根据本次姓名生成问候', startsRound: true,
            run: state => ({ answer: '你好，' + state.name }),
            routes: [route.deliver('answer')],
        }],
});
const agent = new AgentCoreEngine({ workflow, execution: {
        observe: { event(event) { console.log(event.type, event.eventId); } },
    } });
console.log(await agent.run({ name: 'Ivan' }));

Report a business stage while the node executes

Configuration · Node.js

The node only reports facts it knows; Core owns node identity and execution association.

TypeScript

import { AgentCoreEngine, defineWorkflow, route } from '@jangts/agent-core/engine';
type State = { name: string; answer?: string };
const workflow = defineWorkflow<State, string>({ id: 'greeting', nodes: [{
  id: 'greet', description: '生成问候', startsRound: true,
  run(state, context) {
    context.report({ payload: { phase: 'formatting' } });
    return { answer: '你好，' + state.name };
  }, routes: [route.deliver('answer')],
}] });
const agent = new AgentCoreEngine({ workflow, execution: {
  observe: { event(event) { console.log(event.type); } },
} });
console.log(await agent.run({ name: 'Ivan' }));

JavaScript

import { AgentCoreEngine, defineWorkflow, route } from '@jangts/agent-core/engine';
const workflow = defineWorkflow({ id: 'greeting', nodes: [{
            id: 'greet', description: '生成问候', startsRound: true,
            run(state, context) {
                context.report({ payload: { phase: 'formatting' } });
                return { answer: '你好，' + state.name };
            }, routes: [route.deliver('answer')],
        }] });
const agent = new AgentCoreEngine({ workflow, execution: {
        observe: { event(event) { console.log(event.type); } },
    } });
console.log(await agent.run({ name: 'Ivan' }));

Record an async plan first, then fill in the result

Configuration · Node.js

A scoped logger binds the group and only fixes its own writes; readAll exposes the full ledger. Every fix appends an independent record linked to its original write, without logger submission.

TypeScript

import { AgentCoreEngine, defineWorkflow, route } from '@jangts/agent-core/engine';
type State = { result?: string };
const workflow = defineWorkflow<State, string>({ id: 'prepare', nodes: [{
  id: 'prepare', description: '准备资料', async run(_, context) {
    const subs = context.createSubFlowLogger();
    const read = subs.fork('material-read');
    const id = read.write({ name: '读取资料', kind: 'read', type: 'material.read' });
    read.fix(id, { status: 'running' });
    const result = await Promise.resolve('已准备');
    read.fix(id, { status: 'success', usedModel: false, payload: { result } });
    context.report({ usedModel: false });
    console.log(read.readAll());
    return { result };
  }, routes: [route.deliver('result')],
}] });
const agent = new AgentCoreEngine({ workflow, execution: { observe: {
  accepts: ({ event, content }) => content === 'metadata' || event === 'node.progress',
  event(event) { if (event.type === 'node.progress') console.log(event.payload.subStep, event.subSteps); },
} } });
console.log(await agent.run({}));

JavaScript

import { AgentCoreEngine, defineWorkflow, route } from '@jangts/agent-core/engine';
const workflow = defineWorkflow({ id: 'prepare', nodes: [{
            id: 'prepare', description: '准备资料', async run(_, context) {
                const subs = context.createSubFlowLogger();
                const read = subs.fork('material-read');
                const id = read.write({ name: '读取资料', kind: 'read', type: 'material.read' });
                read.fix(id, { status: 'running' });
                const result = await Promise.resolve('已准备');
                read.fix(id, { status: 'success', usedModel: false, payload: { result } });
                context.report({ usedModel: false });
                console.log(read.readAll());
                return { result };
            }, routes: [route.deliver('result')],
        }] });
const agent = new AgentCoreEngine({ workflow, execution: { observe: {
            accepts: ({ event, content }) => content === 'metadata' || event === 'node.progress',
            event(event) { if (event.type === 'node.progress')
                console.log(event.payload.subStep, event.subSteps); },
        } } });
console.log(await agent.run({}));

Self-orchestrated execution uses the same report context

Configuration · Node.js

write/fix emit progress events directly, so report does not need to be called again for each revision. The default observer receives only metadata.

TypeScript

import { AgentCoreEngine } from '@jangts/agent-core/engine';
const agent = new AgentCoreEngine<string>({ execution: { observe: {
  event(event) {
    if (event.type === 'node.progress') console.log(event.payload.subStep);
  },
} } });
const answer = await agent.run(async (_, context) => {
  context.report({ usedModel: false });
  const subs = context.createSubFlowLogger();
  const id = subs.write({ name: '整理结果', kind: 'format', status: 'running' });
  const value = await Promise.resolve('已完成');
  subs.fix(id, { status: 'success', usedModel: false });
  return { type: 'complete', value };
});
console.log(answer);

JavaScript

import { AgentCoreEngine } from '@jangts/agent-core/engine';
const agent = new AgentCoreEngine({ execution: { observe: {
            event(event) {
                if (event.type === 'node.progress')
                    console.log(event.payload.subStep);
            },
        } } });
const answer = await agent.run(async (_, context) => {
    context.report({ usedModel: false });
    const subs = context.createSubFlowLogger();
    const id = subs.write({ name: '整理结果', kind: 'format', status: 'running' });
    const value = await Promise.resolve('已完成');
    subs.fix(id, { status: 'success', usedModel: false });
    return { type: 'complete', value };
});
console.log(answer);

Policies & implementations

• Shared execution fact source and consumption isolation
• Sub-step statuses and their progression