mirror of
https://github.com/n8n-io/n8n.git
synced 2026-07-27 11:05:14 +02:00
refactor(core): Drop redundant agent execution columns (#32862)
This commit is contained in:
yehorkardash
GitHub
parent
2e6bc87838
commit
bc3fe3eaa6
|
|
@ -10,7 +10,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand.
|
|||
| ---- | ------- | ------- | ---- |
|
||||
| [public.agent_chat_subscriptions](public.agent_chat_subscriptions.md) | 6 | | BASE TABLE |
|
||||
| [public.agent_checkpoints](public.agent_checkpoints.md) | 6 | | BASE TABLE |
|
||||
| [public.agent_execution](public.agent_execution.md) | 20 | | BASE TABLE |
|
||||
| [public.agent_execution](public.agent_execution.md) | 18 | | BASE TABLE |
|
||||
| [public.agent_execution_threads](public.agent_execution_threads.md) | 17 | | BASE TABLE |
|
||||
| [public.agent_files](public.agent_files.md) | 8 | | BASE TABLE |
|
||||
| [public.agent_history](public.agent_history.md) | 9 | | BASE TABLE |
|
||||
|
|
@ -307,7 +307,6 @@ erDiagram
|
|||
timestamp_3__with_time_zone updatedAt
|
||||
}
|
||||
"public.agent_execution" {
|
||||
text assistantResponse
|
||||
integer completionTokens
|
||||
double_precision cost
|
||||
timestamp_3__with_time_zone createdAt
|
||||
|
|
@ -323,7 +322,6 @@ erDiagram
|
|||
timestamp_3__with_time_zone stoppedAt
|
||||
varchar_128_ threadId FK
|
||||
json timeline
|
||||
json toolCalls
|
||||
integer totalTokens
|
||||
timestamp_3__with_time_zone updatedAt
|
||||
text userMessage
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@
|
|||
|
||||
| Name | Type | Default | Nullable | Children | Parents | Comment |
|
||||
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
|
||||
| assistantResponse | text | | false | | | |
|
||||
| completionTokens | integer | | true | | | |
|
||||
| cost | double precision | | true | | | |
|
||||
| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
|
||||
|
|
@ -20,10 +19,9 @@
|
|||
| stoppedAt | timestamp(3) with time zone | | true | | | |
|
||||
| threadId | varchar(128) | | false | | [public.agent_execution_threads](public.agent_execution_threads.md) | |
|
||||
| timeline | json | | true | | | |
|
||||
| toolCalls | json | | true | | | |
|
||||
| totalTokens | integer | | true | | | |
|
||||
| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | |
|
||||
| userMessage | text | | false | | | |
|
||||
| userMessage | text | | true | | | |
|
||||
|
||||
## Constraints
|
||||
|
||||
|
|
@ -33,14 +31,12 @@
|
|||
| CHK_agent_execution_status | CHECK | CHECK (((status)::text = ANY ((ARRAY['success'::character varying, 'error'::character varying])::text[]))) |
|
||||
| FK_add2432fb6034cc18b6af299dce | FOREIGN KEY | FOREIGN KEY ("threadId") REFERENCES agent_execution_threads(id) ON DELETE CASCADE |
|
||||
| PK_ba438acc8532addc12d1ef17049 | PRIMARY KEY | PRIMARY KEY (id) |
|
||||
| agent_execution_assistantResponse_not_null | n | NOT NULL "assistantResponse" |
|
||||
| agent_execution_createdAt_not_null | n | NOT NULL "createdAt" |
|
||||
| agent_execution_duration_not_null | n | NOT NULL duration |
|
||||
| agent_execution_id_not_null | n | NOT NULL id |
|
||||
| agent_execution_status_not_null | n | NOT NULL status |
|
||||
| agent_execution_threadId_not_null | n | NOT NULL "threadId" |
|
||||
| agent_execution_updatedAt_not_null | n | NOT NULL "updatedAt" |
|
||||
| agent_execution_userMessage_not_null | n | NOT NULL "userMessage" |
|
||||
|
||||
## Indexes
|
||||
|
||||
|
|
@ -57,7 +53,6 @@ erDiagram
|
|||
"public.agent_execution" }o--|| "public.agent_execution_threads" : "FOREIGN KEY (#quot;threadId#quot;) REFERENCES agent_execution_threads(id) ON DELETE CASCADE"
|
||||
|
||||
"public.agent_execution" {
|
||||
text assistantResponse
|
||||
integer completionTokens
|
||||
double_precision cost
|
||||
timestamp_3__with_time_zone createdAt
|
||||
|
|
@ -73,7 +68,6 @@ erDiagram
|
|||
timestamp_3__with_time_zone stoppedAt
|
||||
varchar_128_ threadId FK
|
||||
json timeline
|
||||
json toolCalls
|
||||
integer totalTokens
|
||||
timestamp_3__with_time_zone updatedAt
|
||||
text userMessage
|
||||
|
|
|
|||
|
|
@ -94,7 +94,6 @@ erDiagram
|
|||
varchar_36_ versionId
|
||||
}
|
||||
"public.agent_execution" {
|
||||
text assistantResponse
|
||||
integer completionTokens
|
||||
double_precision cost
|
||||
timestamp_3__with_time_zone createdAt
|
||||
|
|
@ -110,7 +109,6 @@ erDiagram
|
|||
timestamp_3__with_time_zone stoppedAt
|
||||
varchar_128_ threadId FK
|
||||
json timeline
|
||||
json toolCalls
|
||||
integer totalTokens
|
||||
timestamp_3__with_time_zone updatedAt
|
||||
text userMessage
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand.
|
|||
| ---- | ------- | ------- | ---- |
|
||||
| [agent_chat_subscriptions](agent_chat_subscriptions.md) | 6 | | table |
|
||||
| [agent_checkpoints](agent_checkpoints.md) | 6 | | table |
|
||||
| [agent_execution](agent_execution.md) | 20 | | table |
|
||||
| [agent_execution](agent_execution.md) | 18 | | table |
|
||||
| [agent_execution_threads](agent_execution_threads.md) | 17 | | table |
|
||||
| [agent_files](agent_files.md) | 8 | | table |
|
||||
| [agent_history](agent_history.md) | 9 | | table |
|
||||
|
|
@ -295,7 +295,6 @@ erDiagram
|
|||
datetime_3_ updatedAt
|
||||
}
|
||||
"agent_execution" {
|
||||
TEXT assistantResponse
|
||||
INTEGER completionTokens
|
||||
REAL cost
|
||||
datetime_3_ createdAt
|
||||
|
|
@ -311,7 +310,6 @@ erDiagram
|
|||
datetime_3_ stoppedAt
|
||||
varchar_128_ threadId FK
|
||||
TEXT timeline
|
||||
TEXT toolCalls
|
||||
INTEGER totalTokens
|
||||
datetime_3_ updatedAt
|
||||
TEXT userMessage
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@
|
|||
<summary><strong>Table Definition</strong></summary>
|
||||
|
||||
```sql
|
||||
CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId" varchar(128) NOT NULL, "status" varchar(16) NOT NULL, "startedAt" datetime(3), "stoppedAt" datetime(3), "duration" integer NOT NULL DEFAULT (0), "userMessage" text NOT NULL, "assistantResponse" text NOT NULL, "model" varchar(255), "promptTokens" integer, "completionTokens" integer, "totalTokens" integer, "cost" real, "toolCalls" text, "timeline" text, "error" text, "hitlStatus" varchar(16), "source" varchar(32), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "CHK_agent_execution_status" CHECK (("status" IN ('success', 'error'))), CONSTRAINT "CHK_agent_execution_hitlStatus" CHECK (("hitlStatus" IN ('suspended', 'resumed'))), CONSTRAINT "FK_add2432fb6034cc18b6af299dce" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION)
|
||||
CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId" varchar(128) NOT NULL, "status" varchar(16) NOT NULL, "startedAt" datetime(3), "stoppedAt" datetime(3), "duration" integer NOT NULL DEFAULT (0), "userMessage" text, "model" varchar(255), "promptTokens" integer, "completionTokens" integer, "totalTokens" integer, "cost" real, "timeline" text, "error" text, "hitlStatus" varchar(16), "source" varchar(32), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "CHK_agent_execution_status" CHECK (((("status" IN ('success', 'error'))))), CONSTRAINT "CHK_agent_execution_hitlStatus" CHECK (((("hitlStatus" IN ('suspended', 'resumed'))))), CONSTRAINT "FK_add2432fb6034cc18b6af299dce" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION)
|
||||
```
|
||||
|
||||
</details>
|
||||
|
|
@ -15,7 +15,6 @@ CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId
|
|||
|
||||
| Name | Type | Default | Nullable | Children | Parents | Comment |
|
||||
| ---- | ---- | ------- | -------- | -------- | ------- | ------- |
|
||||
| assistantResponse | TEXT | | false | | | |
|
||||
| completionTokens | INTEGER | | true | | | |
|
||||
| cost | REAL | | true | | | |
|
||||
| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
|
||||
|
|
@ -31,17 +30,16 @@ CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId
|
|||
| stoppedAt | datetime(3) | | true | | | |
|
||||
| threadId | varchar(128) | | false | | [agent_execution_threads](agent_execution_threads.md) | |
|
||||
| timeline | TEXT | | true | | | |
|
||||
| toolCalls | TEXT | | true | | | |
|
||||
| totalTokens | INTEGER | | true | | | |
|
||||
| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | |
|
||||
| userMessage | TEXT | | false | | | |
|
||||
| userMessage | TEXT | | true | | | |
|
||||
|
||||
## Constraints
|
||||
|
||||
| Name | Type | Definition |
|
||||
| ---- | ---- | ---------- |
|
||||
| - | CHECK | CHECK (("status" IN ('success', 'error'))) |
|
||||
| - | CHECK | CHECK (("hitlStatus" IN ('suspended', 'resumed'))) |
|
||||
| - | CHECK | CHECK (((("status" IN ('success', 'error'))))) |
|
||||
| - | CHECK | CHECK (((("hitlStatus" IN ('suspended', 'resumed'))))) |
|
||||
| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
|
||||
| id | PRIMARY KEY | PRIMARY KEY (id) |
|
||||
| sqlite_autoindex_agent_execution_1 | PRIMARY KEY | PRIMARY KEY (id) |
|
||||
|
|
@ -61,7 +59,6 @@ erDiagram
|
|||
"agent_execution" }o--|| "agent_execution_threads" : "FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE"
|
||||
|
||||
"agent_execution" {
|
||||
TEXT assistantResponse
|
||||
INTEGER completionTokens
|
||||
REAL cost
|
||||
datetime_3_ createdAt
|
||||
|
|
@ -77,7 +74,6 @@ erDiagram
|
|||
datetime_3_ stoppedAt
|
||||
varchar_128_ threadId FK
|
||||
TEXT timeline
|
||||
TEXT toolCalls
|
||||
INTEGER totalTokens
|
||||
datetime_3_ updatedAt
|
||||
TEXT userMessage
|
||||
|
|
|
|||
|
|
@ -95,7 +95,6 @@ erDiagram
|
|||
varchar_36_ versionId
|
||||
}
|
||||
"agent_execution" {
|
||||
TEXT assistantResponse
|
||||
INTEGER completionTokens
|
||||
REAL cost
|
||||
datetime_3_ createdAt
|
||||
|
|
@ -111,7 +110,6 @@ erDiagram
|
|||
datetime_3_ stoppedAt
|
||||
varchar_128_ threadId FK
|
||||
TEXT timeline
|
||||
TEXT toolCalls
|
||||
INTEGER totalTokens
|
||||
datetime_3_ updatedAt
|
||||
TEXT userMessage
|
||||
|
|
|
|||
|
|
@ -0,0 +1,28 @@
|
|||
import type { MigrationContext, ReversibleMigration } from '../migration-types';
|
||||
|
||||
export class DropAgentExecutionFallbackColumns1784000000039 implements ReversibleMigration {
|
||||
async up({ schemaBuilder: { dropColumns, dropNotNull } }: MigrationContext) {
|
||||
await dropColumns('agent_execution', ['assistantResponse', 'toolCalls'], {
|
||||
recreatesOnSqlite: true,
|
||||
});
|
||||
await dropNotNull('agent_execution', 'userMessage', { recreatesOnSqlite: true });
|
||||
}
|
||||
|
||||
async down(ctx: MigrationContext) {
|
||||
const {
|
||||
escape,
|
||||
schemaBuilder: { addColumns, addNotNull, column },
|
||||
} = ctx;
|
||||
|
||||
await addColumns(
|
||||
'agent_execution',
|
||||
[column('assistantResponse').text.notNull.default("''"), column('toolCalls').json],
|
||||
{ recreatesOnSqlite: true },
|
||||
);
|
||||
|
||||
await ctx.runQuery(
|
||||
`UPDATE ${escape.tableName('agent_execution')} SET ${escape.columnName('userMessage')} = '' WHERE ${escape.columnName('userMessage')} IS NULL`,
|
||||
);
|
||||
await addNotNull('agent_execution', 'userMessage', { recreatesOnSqlite: true });
|
||||
}
|
||||
}
|
||||
|
|
@ -212,6 +212,7 @@ import { AddUniqueAgentFileNames1784000000035 } from '../common/1784000000035-Ad
|
|||
import { CreateInstanceAiThreadGrantTable1784000000036 } from '../common/1784000000036-CreateInstanceAiThreadGrantTable';
|
||||
import { DropAgentDescriptionFromAgents1784000000037 } from '../common/1784000000037-DropAgentDescriptionFromAgents';
|
||||
import { SetChatHubEnabledFromUsage1784000000038 } from '../common/1784000000038-SetChatHubEnabledFromUsage';
|
||||
import { DropAgentExecutionFallbackColumns1784000000039 } from '../common/1784000000039-DropAgentExecutionFallbackColumns';
|
||||
import type { Migration } from '../migration-types';
|
||||
|
||||
export const postgresMigrations: Migration[] = [
|
||||
|
|
@ -429,4 +430,5 @@ export const postgresMigrations: Migration[] = [
|
|||
CreateInstanceAiThreadGrantTable1784000000036,
|
||||
DropAgentDescriptionFromAgents1784000000037,
|
||||
SetChatHubEnabledFromUsage1784000000038,
|
||||
DropAgentExecutionFallbackColumns1784000000039,
|
||||
];
|
||||
|
|
|
|||
|
|
@ -205,6 +205,7 @@ import { AddBinaryDataSizeBytesToExecutionEntity1784000000033 } from '../common/
|
|||
import { AddUniqueAgentFileNames1784000000035 } from '../common/1784000000035-AddUniqueAgentFileNames';
|
||||
import { CreateInstanceAiThreadGrantTable1784000000036 } from '../common/1784000000036-CreateInstanceAiThreadGrantTable';
|
||||
import { SetChatHubEnabledFromUsage1784000000038 } from '../common/1784000000038-SetChatHubEnabledFromUsage';
|
||||
import { DropAgentExecutionFallbackColumns1784000000039 } from '../common/1784000000039-DropAgentExecutionFallbackColumns';
|
||||
|
||||
const sqliteMigrations: Migration[] = [
|
||||
InitialMigration1588102412422,
|
||||
|
|
@ -413,6 +414,7 @@ const sqliteMigrations: Migration[] = [
|
|||
CreateInstanceAiThreadGrantTable1784000000036,
|
||||
DropAgentDescriptionFromAgents1784000000037,
|
||||
SetChatHubEnabledFromUsage1784000000038,
|
||||
DropAgentExecutionFallbackColumns1784000000039,
|
||||
];
|
||||
|
||||
export { sqliteMigrations };
|
||||
|
|
|
|||
|
|
@ -375,7 +375,13 @@ describe('AgentExecutionOrchestratorService', () => {
|
|||
const { service, executionService } = makeService();
|
||||
executionService.getThreadDetail.mockResolvedValue({
|
||||
thread: { id: 'thread-1' },
|
||||
executions: [{ id: 'execution-1', userMessage: 'Hi', assistantResponse: 'Hello' }],
|
||||
executions: [
|
||||
{
|
||||
id: 'execution-1',
|
||||
userMessage: 'Hi',
|
||||
timeline: [{ type: 'text', content: 'Hello', timestamp: 100 }],
|
||||
},
|
||||
],
|
||||
} as never);
|
||||
|
||||
await expect(
|
||||
|
|
@ -433,7 +439,7 @@ describe('AgentExecutionOrchestratorService', () => {
|
|||
expect(executionService.recordMessage).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
threadId: 'thread-1',
|
||||
userMessage: '',
|
||||
userMessage: null,
|
||||
hitlStatus: 'resumed',
|
||||
telemetry: {
|
||||
runType: 'production',
|
||||
|
|
@ -472,7 +478,7 @@ describe('AgentExecutionOrchestratorService', () => {
|
|||
);
|
||||
|
||||
expect(executionService.recordMessage).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ threadId: 'thread-1', userMessage: '', hitlStatus: 'suspended' }),
|
||||
expect.objectContaining({ threadId: 'thread-1', userMessage: null, hitlStatus: 'suspended' }),
|
||||
);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -42,7 +42,6 @@ function makeMessageRecord(overrides: Partial<MessageRecord> = {}): MessageRecor
|
|||
finishReason: 'stop',
|
||||
usage: null,
|
||||
totalCost: null,
|
||||
toolCalls: [],
|
||||
timeline: [],
|
||||
startTime: 0,
|
||||
duration: 1,
|
||||
|
|
@ -87,7 +86,6 @@ describe('AgentExecutionService', () => {
|
|||
finishReason: 'stop',
|
||||
usage: { promptTokens: 10, completionTokens: 5, totalTokens: 15 },
|
||||
totalCost: 0.01,
|
||||
toolCalls: [],
|
||||
timeline: [],
|
||||
startTime: Date.parse('2026-05-07T10:00:00Z'),
|
||||
duration: 1234,
|
||||
|
|
@ -223,7 +221,19 @@ describe('AgentExecutionService', () => {
|
|||
record: makeMessageRecord({
|
||||
usage: { promptTokens: 10, completionTokens: 5, totalTokens: 15 },
|
||||
totalCost: 25,
|
||||
toolCalls: [{ name: 'lookup', input: {}, output: {} }],
|
||||
timeline: [
|
||||
{
|
||||
type: 'tool-call',
|
||||
kind: 'tool',
|
||||
name: 'lookup',
|
||||
toolCallId: 'tc1',
|
||||
input: {},
|
||||
output: {},
|
||||
startTime: 0,
|
||||
endTime: 123,
|
||||
success: true,
|
||||
},
|
||||
],
|
||||
duration: 123,
|
||||
}),
|
||||
telemetry: {
|
||||
|
|
|
|||
|
|
@ -175,50 +175,8 @@ describe('ExecutionRecorder', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('backward compat', () => {
|
||||
it('still populates flat toolCalls array', () => {
|
||||
const recorder = new ExecutionRecorder();
|
||||
|
||||
recorder.record(makeToolCallChunk('my_tool', { x: 1 }));
|
||||
recorder.record(makeToolResultChunk('my_tool', { y: 2 }));
|
||||
recorder.record({ type: 'finish', finishReason: 'stop' } as StreamChunk);
|
||||
|
||||
const record = recorder.getMessageRecord();
|
||||
|
||||
expect(record.toolCalls).toHaveLength(1);
|
||||
expect(record.toolCalls[0]).toEqual({
|
||||
name: 'my_tool',
|
||||
input: { x: 1 },
|
||||
output: { y: 2 },
|
||||
});
|
||||
});
|
||||
|
||||
it('pairs same-name flat tool calls by toolCallId when results arrive out of order', () => {
|
||||
const recorder = new ExecutionRecorder();
|
||||
|
||||
recorder.record(makeToolCallChunk('same_name_tool', { file: 'first' }, 'call-1'));
|
||||
recorder.record(makeToolCallChunk('same_name_tool', { file: 'second' }, 'call-2'));
|
||||
recorder.record(makeToolResultChunk('same_name_tool', { result: 'second' }, 'call-2'));
|
||||
recorder.record(makeToolResultChunk('same_name_tool', { result: 'first' }, 'call-1'));
|
||||
recorder.record({ type: 'finish', finishReason: 'stop' } as StreamChunk);
|
||||
|
||||
const record = recorder.getMessageRecord();
|
||||
|
||||
expect(record.toolCalls).toEqual([
|
||||
{
|
||||
name: 'same_name_tool',
|
||||
input: { file: 'first' },
|
||||
output: { result: 'first' },
|
||||
},
|
||||
{
|
||||
name: 'same_name_tool',
|
||||
input: { file: 'second' },
|
||||
output: { result: 'second' },
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('still concatenates assistantResponse from all text deltas', () => {
|
||||
describe('message record', () => {
|
||||
it('concatenates assistantResponse from all text deltas', () => {
|
||||
const recorder = new ExecutionRecorder();
|
||||
|
||||
recorder.record({ type: 'text-delta', id: 't1', delta: 'Hello ' });
|
||||
|
|
@ -233,7 +191,7 @@ describe('ExecutionRecorder', () => {
|
|||
});
|
||||
|
||||
describe('secret scrubbing', () => {
|
||||
it('sanitizes tool inputs and outputs in flat records and timeline entries', () => {
|
||||
it('sanitizes tool inputs and outputs in timeline entries', () => {
|
||||
const recorder = new ExecutionRecorder();
|
||||
|
||||
recorder.record(
|
||||
|
|
@ -253,18 +211,6 @@ describe('ExecutionRecorder', () => {
|
|||
const record = recorder.getMessageRecord();
|
||||
const timelineEntry = record.timeline.find((e) => e.type === 'tool-call');
|
||||
|
||||
expect(record.toolCalls[0]).toEqual({
|
||||
name: 'lookup',
|
||||
input: {
|
||||
query: 'project status',
|
||||
password: '[REDACTED]',
|
||||
nested: { apiKey: '[REDACTED]' },
|
||||
},
|
||||
output: {
|
||||
result: '[REDACTED]',
|
||||
authorization: '[REDACTED]',
|
||||
},
|
||||
});
|
||||
expect(timelineEntry).toMatchObject({
|
||||
input: {
|
||||
query: 'project status',
|
||||
|
|
@ -291,8 +237,9 @@ describe('ExecutionRecorder', () => {
|
|||
});
|
||||
|
||||
const record = recorder.getMessageRecord();
|
||||
const timelineEntry = record.timeline.find((e) => e.type === 'tool-call');
|
||||
|
||||
expect(record.toolCalls[0].output).toEqual({ error: '[REDACTED]' });
|
||||
expect(timelineEntry).toMatchObject({ output: { error: '[REDACTED]' } });
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -570,12 +517,6 @@ describe('ExecutionRecorder — workflow-tool timeline tags', () => {
|
|||
expect(tc?.workflowName).toBe('Run WF');
|
||||
expect(tc?.workflowExecutionId).toBe('e-99');
|
||||
expect(tc?.success).toBe(true);
|
||||
expect(record.toolCalls).toHaveLength(1);
|
||||
expect(record.toolCalls[0]).toEqual({
|
||||
name: 'run-wf',
|
||||
input: undefined,
|
||||
output: { executionId: 'e-99', status: 'success' },
|
||||
});
|
||||
});
|
||||
|
||||
it('leaves workflowExecutionId undefined when the output is an error with no executionId', () => {
|
||||
|
|
|
|||
|
|
@ -8,10 +8,7 @@ function execution(overrides: Partial<AgentExecution> = {}): AgentExecution {
|
|||
return {
|
||||
id: 'execution-1',
|
||||
userMessage: 'Hello',
|
||||
assistantResponse: '',
|
||||
toolCalls: null,
|
||||
timeline: null,
|
||||
error: null,
|
||||
...overrides,
|
||||
} as unknown as AgentExecution;
|
||||
}
|
||||
|
|
@ -20,7 +17,6 @@ describe('execution-to-message-mapper', () => {
|
|||
it('maps execution timeline text and tool calls into assistant message content', () => {
|
||||
const result = executionToMessagesDto(
|
||||
execution({
|
||||
assistantResponse: 'Let me check.Done.',
|
||||
timeline: [
|
||||
{ type: 'text', content: 'Let me check.', timestamp: 100, endTime: 110 },
|
||||
{
|
||||
|
|
@ -112,76 +108,18 @@ describe('execution-to-message-mapper', () => {
|
|||
]);
|
||||
});
|
||||
|
||||
it('falls back to recorded tool calls when execution timeline is unavailable', () => {
|
||||
const result = executionToMessagesDto(
|
||||
execution({
|
||||
assistantResponse: 'Legacy done.',
|
||||
toolCalls: [{ name: 'legacy_tool', input: { id: '123' }, output: 'ok' }],
|
||||
}),
|
||||
);
|
||||
|
||||
expect(result).toEqual([
|
||||
{
|
||||
id: 'execution-1:user',
|
||||
role: 'user',
|
||||
content: [{ type: 'text', text: 'Hello' }],
|
||||
},
|
||||
{
|
||||
id: 'execution-1:assistant',
|
||||
role: 'assistant',
|
||||
content: [
|
||||
{
|
||||
type: 'tool-call',
|
||||
toolName: 'legacy_tool',
|
||||
toolCallId: 'execution-1:tool:0',
|
||||
input: { id: '123' },
|
||||
output: 'ok',
|
||||
},
|
||||
{ type: 'text', text: 'Legacy done.' },
|
||||
],
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('does not infer resolved state from legacy recorded tool call output', () => {
|
||||
const result = executionToMessagesDto(
|
||||
execution({
|
||||
toolCalls: [
|
||||
{
|
||||
name: 'legacy_tool',
|
||||
input: { id: '123' },
|
||||
output: { message: 'Tool failed before timeline recording was available' },
|
||||
},
|
||||
],
|
||||
}),
|
||||
);
|
||||
|
||||
expect(result).toEqual([
|
||||
{
|
||||
id: 'execution-1:user',
|
||||
role: 'user',
|
||||
content: [{ type: 'text', text: 'Hello' }],
|
||||
},
|
||||
{
|
||||
id: 'execution-1:assistant',
|
||||
role: 'assistant',
|
||||
content: [
|
||||
{
|
||||
type: 'tool-call',
|
||||
toolName: 'legacy_tool',
|
||||
toolCallId: 'execution-1:tool:0',
|
||||
input: { id: '123' },
|
||||
output: { message: 'Tool failed before timeline recording was available' },
|
||||
},
|
||||
],
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('flattens multiple executions into a single message list', () => {
|
||||
const result = executionsToMessagesDto([
|
||||
execution({ id: 'execution-1', userMessage: 'Hello', assistantResponse: 'Hi' }),
|
||||
execution({ id: 'execution-2', userMessage: 'Again', assistantResponse: 'There' }),
|
||||
execution({
|
||||
id: 'execution-1',
|
||||
userMessage: 'Hello',
|
||||
timeline: [{ type: 'text', content: 'Hi', timestamp: 100 }],
|
||||
}),
|
||||
execution({
|
||||
id: 'execution-2',
|
||||
userMessage: 'Again',
|
||||
timeline: [{ type: 'text', content: 'There', timestamp: 200 }],
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(result.map((message) => message.id)).toEqual([
|
||||
|
|
@ -224,7 +162,7 @@ describe('execution-to-message-mapper', () => {
|
|||
}),
|
||||
execution({
|
||||
id: 'execution-resumed',
|
||||
userMessage: '',
|
||||
userMessage: null,
|
||||
timeline: [
|
||||
{
|
||||
type: 'tool-call',
|
||||
|
|
|
|||
|
|
@ -280,7 +280,7 @@ export class AgentExecutionOrchestratorService {
|
|||
agentId,
|
||||
agentName: agentInstance.name,
|
||||
projectId,
|
||||
userMessage: '',
|
||||
userMessage: null,
|
||||
record: messageRecord,
|
||||
hitlStatus: recorder.suspended ? 'suspended' : 'resumed',
|
||||
telemetry: {
|
||||
|
|
|
|||
|
|
@ -17,7 +17,7 @@ export interface RecordMessageParams {
|
|||
agentId: string;
|
||||
agentName: string;
|
||||
projectId: string;
|
||||
userMessage: string;
|
||||
userMessage: string | null;
|
||||
record: MessageRecord;
|
||||
/** Set to 'suspended' or 'resumed' for HITL tool call flows. */
|
||||
hitlStatus?: 'suspended' | 'resumed';
|
||||
|
|
@ -41,7 +41,7 @@ export interface ThreadDetail {
|
|||
executions: AgentExecution[];
|
||||
}
|
||||
|
||||
export interface ThreadListItem extends AgentExecutionThread {
|
||||
export interface ThreadListItem extends Omit<AgentExecutionThread, 'generateId' | 'setUpdateDate'> {
|
||||
firstMessage: string | null;
|
||||
}
|
||||
|
||||
|
|
@ -92,10 +92,16 @@ export class AgentExecutionService {
|
|||
}
|
||||
|
||||
// Replace platform mentions (e.g. Slack's <@U0ANB4K6611> or plain @U0ANB4K6611)
|
||||
const cleanedMessage = params.userMessage
|
||||
.replace(/<@[A-Z0-9]+>/gi, `@${agentName}`)
|
||||
.replace(/@[A-Z0-9]{8,}/gi, `@${agentName}`)
|
||||
.trim();
|
||||
const userMessage =
|
||||
params.userMessage === null
|
||||
? null
|
||||
: (() => {
|
||||
const cleanedMessage = params.userMessage
|
||||
.replace(/<@[A-Z0-9]+>/gi, `@${agentName}`)
|
||||
.replace(/@[A-Z0-9]{8,}/gi, `@${agentName}`)
|
||||
.trim();
|
||||
return cleanedMessage.length > 0 ? cleanedMessage : null;
|
||||
})();
|
||||
|
||||
const status: AgentExecution['status'] = record.error ? 'error' : 'success';
|
||||
const startedAt = new Date(record.startTime);
|
||||
|
|
@ -108,14 +114,12 @@ export class AgentExecutionService {
|
|||
startedAt,
|
||||
stoppedAt,
|
||||
duration: record.duration,
|
||||
userMessage: cleanedMessage,
|
||||
assistantResponse: record.assistantResponse,
|
||||
userMessage,
|
||||
model: record.model,
|
||||
promptTokens: record.usage?.promptTokens ?? null,
|
||||
completionTokens: record.usage?.completionTokens ?? null,
|
||||
totalTokens: record.usage?.totalTokens ?? null,
|
||||
cost: record.totalCost,
|
||||
toolCalls: record.toolCalls.length > 0 ? record.toolCalls : null,
|
||||
timeline: record.timeline.length > 0 ? record.timeline : null,
|
||||
error: record.error,
|
||||
hitlStatus: hitlStatus ?? null,
|
||||
|
|
@ -151,7 +155,7 @@ export class AgentExecutionService {
|
|||
configuration: params.telemetry.configuration,
|
||||
latency_ms: record.duration,
|
||||
cost: record.totalCost ?? 0,
|
||||
tool_call_count: record.toolCalls.length,
|
||||
tool_call_count: record.timeline.filter((t) => t.type === 'tool-call').length,
|
||||
});
|
||||
} catch (error) {
|
||||
this.logger.warn('Failed to track agent execution telemetry', {
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import { DateTimeColumn, JsonColumn, WithTimestampsAndStringId } from '@n8n/db';
|
|||
import { Column, Entity, Index, JoinColumn, ManyToOne } from '@n8n/typeorm';
|
||||
|
||||
import { AgentExecutionThread } from './agent-execution-thread.entity';
|
||||
import type { RecordedToolCall, TimelineEvent } from '../execution-recorder';
|
||||
import type { TimelineEvent } from '../execution-recorder';
|
||||
|
||||
export type AgentExecutionStatus = 'success' | 'error';
|
||||
export type AgentExecutionHitlStatus = 'suspended' | 'resumed';
|
||||
|
|
@ -44,15 +44,9 @@ export class AgentExecution extends WithTimestampsAndStringId {
|
|||
@Column({ type: 'int', default: 0 })
|
||||
duration: number;
|
||||
|
||||
/**
|
||||
* Cleaned user input. Empty for resumed runs (HITL continuations) where
|
||||
* the user input belongs to an earlier suspended run in the same thread.
|
||||
*/
|
||||
@Column({ type: 'text' })
|
||||
userMessage: string;
|
||||
|
||||
@Column({ type: 'text' })
|
||||
assistantResponse: string;
|
||||
/** Cleaned user input. Null for resumed runs where the input belongs to an earlier run. */
|
||||
@Column({ type: 'text', nullable: true })
|
||||
userMessage: string | null;
|
||||
|
||||
@Column({ type: 'varchar', length: 255, nullable: true })
|
||||
model: string | null;
|
||||
|
|
@ -69,9 +63,6 @@ export class AgentExecution extends WithTimestampsAndStringId {
|
|||
@Column({ type: 'double precision', nullable: true })
|
||||
cost: number | null;
|
||||
|
||||
@JsonColumn({ nullable: true })
|
||||
toolCalls: RecordedToolCall[] | null;
|
||||
|
||||
@JsonColumn({ nullable: true })
|
||||
timeline: TimelineEvent[] | null;
|
||||
|
||||
|
|
|
|||
|
|
@ -197,14 +197,6 @@ export interface RecordedUsage {
|
|||
totalTokens: number;
|
||||
}
|
||||
|
||||
export interface RecordedToolCall {
|
||||
name: string;
|
||||
input: unknown;
|
||||
output: unknown;
|
||||
}
|
||||
|
||||
type PendingRecordedToolCall = RecordedToolCall & { toolCallId?: string };
|
||||
|
||||
export type TimelineEvent =
|
||||
| { type: 'text'; content: string; timestamp: number; endTime?: number }
|
||||
| {
|
||||
|
|
@ -244,7 +236,6 @@ export interface MessageRecord {
|
|||
finishReason: string;
|
||||
usage: RecordedUsage | null;
|
||||
totalCost: number | null;
|
||||
toolCalls: RecordedToolCall[];
|
||||
timeline: TimelineEvent[];
|
||||
startTime: number;
|
||||
duration: number;
|
||||
|
|
@ -271,8 +262,6 @@ export class ExecutionRecorder {
|
|||
|
||||
private totalCost: number | null = null;
|
||||
|
||||
private toolCalls: PendingRecordedToolCall[] = [];
|
||||
|
||||
private timeline: TimelineEvent[] = [];
|
||||
|
||||
/** Wall-clock when the first text-delta of the current segment arrived. */
|
||||
|
|
@ -353,7 +342,6 @@ export class ExecutionRecorder {
|
|||
finishReason: this.finishReason,
|
||||
usage: this.usage,
|
||||
totalCost: this.totalCost,
|
||||
toolCalls: this.toolCalls.map(({ toolCallId: _toolCallId, ...toolCall }) => toolCall),
|
||||
timeline: this.timeline,
|
||||
startTime: this.startTime,
|
||||
duration: Date.now() - this.startTime,
|
||||
|
|
@ -381,15 +369,13 @@ export class ExecutionRecorder {
|
|||
}
|
||||
|
||||
/**
|
||||
* Record a discrete `tool-call` chunk from the stream. Maintains both the
|
||||
* flat `toolCalls` array (backward compat) and the ordered timeline. The
|
||||
* matching `tool-result` chunk closes the timeline entry.
|
||||
* Record a discrete `tool-call` chunk from the stream. The matching
|
||||
* `tool-result` chunk closes the timeline entry.
|
||||
*/
|
||||
private recordToolCall(toolCallId: string, name: string, input: unknown): void {
|
||||
this.flushTextBuffer();
|
||||
|
||||
const recordedInput = sanitizeExecutionLogValue(input);
|
||||
this.toolCalls.push({ name, input: recordedInput, output: undefined, toolCallId });
|
||||
|
||||
const entry = this.registry.get(name);
|
||||
// Resolve both `$fromAI(...)` placeholders and simple `={{ $json.x }}`
|
||||
|
|
@ -462,19 +448,6 @@ export class ExecutionRecorder {
|
|||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Find the still-open flat tool-call entry to attach a result to. Prefers
|
||||
* an exact match on `toolCallId`; when the stream omits the id (empty
|
||||
* string), falls back to the most recent open entry (`output === undefined`)
|
||||
* with the same tool name.
|
||||
*/
|
||||
private findOpenToolCall(toolCallId: string, name: string): PendingRecordedToolCall | undefined {
|
||||
if (toolCallId !== '') {
|
||||
return this.toolCalls.find((tc) => tc.toolCallId === toolCallId && tc.output === undefined);
|
||||
}
|
||||
return [...this.toolCalls].reverse().find((tc) => tc.name === name && tc.output === undefined);
|
||||
}
|
||||
|
||||
/**
|
||||
* Record a discrete `tool-result` chunk from the stream. Closes the
|
||||
* matching open timeline entry by `toolCallId` (preferred) or by name as
|
||||
|
|
@ -495,13 +468,6 @@ export class ExecutionRecorder {
|
|||
isError ? normaliseToolErrorOutput(output) : output,
|
||||
);
|
||||
|
||||
const pendingFlat = this.findOpenToolCall(toolCallId, name);
|
||||
if (pendingFlat) {
|
||||
pendingFlat.output = recordedOutput;
|
||||
} else {
|
||||
this.toolCalls.push({ name, input: undefined, output: recordedOutput });
|
||||
}
|
||||
|
||||
const pendingTimeline = [...this.timeline]
|
||||
.reverse()
|
||||
.find(
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ export class AgentExecutionRepository extends Repository<AgentExecution> {
|
|||
* sessions list to render a preview before the LLM-generated title is
|
||||
* available.
|
||||
*
|
||||
* Excludes resumed runs (empty `userMessage`). Returns one row per thread
|
||||
* Excludes resumed runs (null `userMessage`). Returns one row per thread
|
||||
* containing the userMessage from that thread's earliest matching run.
|
||||
*/
|
||||
async findFirstUserMessageByThreadIds(threadIds: string[]): Promise<Map<string, string>> {
|
||||
|
|
@ -33,10 +33,12 @@ export class AgentExecutionRepository extends Repository<AgentExecution> {
|
|||
const rows = await this.createQueryBuilder('e')
|
||||
.select(['e."threadId" AS "threadId"', 'e."userMessage" AS "userMessage"'])
|
||||
.where('e."threadId" IN (:...threadIds)', { threadIds })
|
||||
.andWhere('e."userMessage" IS NOT NULL')
|
||||
.andWhere('e."userMessage" != \'\'')
|
||||
.andWhere(
|
||||
`e."createdAt" = (SELECT MIN(e2."createdAt") FROM ${tableName} e2 ` +
|
||||
'WHERE e2."threadId" = e."threadId" AND e2."userMessage" != \'\')',
|
||||
'WHERE e2."threadId" = e."threadId" AND e2."userMessage" IS NOT NULL ' +
|
||||
'AND e2."userMessage" != \'\')',
|
||||
)
|
||||
.getRawMany<{ threadId: string; userMessage: string }>();
|
||||
|
||||
|
|
|
|||
|
|
@ -2,12 +2,9 @@ import type { AgentPersistedMessageContentPart, AgentPersistedMessageDto } from
|
|||
import { isRecord } from '@n8n/utils';
|
||||
|
||||
import type { AgentExecution } from '../entities/agent-execution.entity';
|
||||
import type { RecordedToolCall, TimelineEvent } from '../execution-recorder';
|
||||
import type { TimelineEvent } from '../execution-recorder';
|
||||
|
||||
type ExecutionTranscript = Pick<
|
||||
AgentExecution,
|
||||
'id' | 'userMessage' | 'assistantResponse' | 'toolCalls' | 'timeline' | 'error'
|
||||
>;
|
||||
type ExecutionTranscript = Pick<AgentExecution, 'id' | 'userMessage' | 'timeline'>;
|
||||
|
||||
type ToolCallTimelineEvent = Extract<TimelineEvent, { type: 'tool-call' }>;
|
||||
type ToolCallContentPart = AgentPersistedMessageContentPart & {
|
||||
|
|
@ -23,8 +20,9 @@ function textPart(text: string): AgentPersistedMessageContentPart | null {
|
|||
function textMessageDto(
|
||||
id: string,
|
||||
role: AgentPersistedMessageDto['role'],
|
||||
text: string,
|
||||
text: string | null,
|
||||
): AgentPersistedMessageDto | null {
|
||||
if (!text) return null;
|
||||
const contentPart = textPart(text);
|
||||
if (!contentPart) return null;
|
||||
|
||||
|
|
@ -115,59 +113,22 @@ function timelineToolCallToPart(event: ToolCallTimelineEvent): AgentPersistedMes
|
|||
};
|
||||
}
|
||||
|
||||
function recordedToolCallToPart(
|
||||
executionId: string,
|
||||
index: number,
|
||||
toolCall: RecordedToolCall,
|
||||
): AgentPersistedMessageContentPart {
|
||||
const base: AgentPersistedMessageContentPart = {
|
||||
type: 'tool-call',
|
||||
toolName: toolCall.name,
|
||||
toolCallId: `${executionId}:tool:${index}`,
|
||||
input: toolCall.input,
|
||||
};
|
||||
|
||||
if (toolCall.output === undefined) return base;
|
||||
|
||||
return {
|
||||
...base,
|
||||
output: toolCall.output,
|
||||
};
|
||||
}
|
||||
|
||||
function assistantContentFromExecution(
|
||||
execution: ExecutionTranscript,
|
||||
): AgentPersistedMessageContentPart[] {
|
||||
const content: AgentPersistedMessageContentPart[] = [];
|
||||
let hasTimelineText = false;
|
||||
let hasTimelineToolCalls = false;
|
||||
|
||||
for (const event of execution.timeline ?? []) {
|
||||
if (event.type === 'text') {
|
||||
const part = textPart(event.content);
|
||||
if (!part) continue;
|
||||
|
||||
hasTimelineText = true;
|
||||
content.push(part);
|
||||
} else if (event.type === 'tool-call') {
|
||||
hasTimelineToolCalls = true;
|
||||
content.push(timelineToolCallToPart(event));
|
||||
}
|
||||
}
|
||||
|
||||
if (!hasTimelineToolCalls) {
|
||||
for (const [index, toolCall] of (execution.toolCalls ?? []).entries()) {
|
||||
content.push(recordedToolCallToPart(execution.id, index, toolCall));
|
||||
}
|
||||
}
|
||||
|
||||
if (!hasTimelineText) {
|
||||
const fallbackText =
|
||||
execution.assistantResponse || (execution.error ? `Error: ${execution.error}` : '');
|
||||
const part = textPart(fallbackText);
|
||||
if (part) content.push(part);
|
||||
}
|
||||
|
||||
return content;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -66,8 +66,7 @@ describe('AgentExecutionRepository', () => {
|
|||
const execution = repository.create({
|
||||
id: uuid(),
|
||||
status: 'success',
|
||||
userMessage: '',
|
||||
assistantResponse: '',
|
||||
userMessage: null,
|
||||
...overrides,
|
||||
} as Partial<AgentExecution>);
|
||||
return await repository.save(execution);
|
||||
|
|
@ -104,12 +103,12 @@ describe('AgentExecutionRepository', () => {
|
|||
expect(result.size).toBe(2);
|
||||
});
|
||||
|
||||
it('skips executions with empty user messages when picking the earliest', async () => {
|
||||
it('skips executions with null user messages when picking the earliest', async () => {
|
||||
const thread = await createThread();
|
||||
|
||||
await createExecution({
|
||||
threadId: thread.id,
|
||||
userMessage: '',
|
||||
userMessage: null,
|
||||
createdAt: new Date('2024-01-01T00:00:00Z'),
|
||||
});
|
||||
await createExecution({
|
||||
|
|
@ -129,12 +128,12 @@ describe('AgentExecutionRepository', () => {
|
|||
expect(result.size).toBe(0);
|
||||
});
|
||||
|
||||
it('omits threads that contain only empty user messages', async () => {
|
||||
it('omits threads that contain only null user messages', async () => {
|
||||
const thread = await createThread();
|
||||
|
||||
await createExecution({
|
||||
threadId: thread.id,
|
||||
userMessage: '',
|
||||
userMessage: null,
|
||||
createdAt: new Date('2024-01-01T00:00:00Z'),
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -182,14 +182,12 @@ function exec(overrides: Partial<AgentExecution> = {}): AgentExecution {
|
|||
startedAt: '2026-04-24T10:00:00Z',
|
||||
stoppedAt: null,
|
||||
duration: 0,
|
||||
userMessage: '',
|
||||
assistantResponse: '',
|
||||
userMessage: null,
|
||||
model: null,
|
||||
promptTokens: null,
|
||||
completionTokens: null,
|
||||
totalTokens: null,
|
||||
cost: null,
|
||||
toolCalls: null,
|
||||
timeline: null,
|
||||
error: null,
|
||||
hitlStatus: null,
|
||||
|
|
|
|||
|
|
@ -33,13 +33,6 @@ export type AgentExecutionHitlStatus = 'suspended' | 'resumed';
|
|||
*/
|
||||
export type AgentExecutionTimelineEvent = Record<string, unknown> & { type: string };
|
||||
|
||||
export interface AgentExecutionToolCall {
|
||||
toolName: string;
|
||||
input: unknown;
|
||||
output: unknown;
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
export interface AgentExecution {
|
||||
id: string;
|
||||
threadId: string;
|
||||
|
|
@ -49,14 +42,12 @@ export interface AgentExecution {
|
|||
startedAt: string | null;
|
||||
stoppedAt: string | null;
|
||||
duration: number;
|
||||
userMessage: string;
|
||||
assistantResponse: string;
|
||||
userMessage: string | null;
|
||||
model: string | null;
|
||||
promptTokens: number | null;
|
||||
completionTokens: number | null;
|
||||
totalTokens: number | null;
|
||||
cost: number | null;
|
||||
toolCalls: AgentExecutionToolCall[] | null;
|
||||
timeline: AgentExecutionTimelineEvent[] | null;
|
||||
error: string | null;
|
||||
hitlStatus: AgentExecutionHitlStatus | null;
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user