wangxuming/LWLT-AIBOT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: bound leadership dashboard filtering

Browse Source
This commit is contained in:
inman
2026-09-02 14:59:18 +08:00
parent d034f649c4
commit 3062ed5f04
8 changed files with 389 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
control-plane/src/server.ts
View File

@@ -186,7 +186,7 @@ const operationsDashboardQuerySchema = z.object({
business_route_id: z.string().trim().max(120).optional(),
status: z.enum(['all', 'active', 'completed', 'attention', 'failed', 'cancelled', 'archived']).default('all'),
search: z.string().trim().max(200).optional(),
limit: z.coerce.number().int().min(1).max(100).default(50),
limit: z.coerce.number().int().min(1).max(100).default(20),
offset: z.coerce.number().int().min(0).max(1_000_000).default(0)
});
@@ -1357,19 +1357,33 @@ export async function buildServer({
setAuthNoStore(reply);
const session = await requireLeadershipSession(request);
const query = operationsDashboardQuerySchema.parse(request.query || {});
return {
ok: true,
...(await tasks.listOperationsDashboard(contextFor(session, request), {
from: query.from,
to: query.to,
actorUserId: query.actor_user_id,
businessRouteId: query.business_route_id,
status: query.status,
search: query.search,
limit: query.limit,
offset: query.offset
}))
const controller = new AbortController();
const abortRequest = () => controller.abort();
const abortClosedReply = () => {
if (!reply.raw.writableEnded) abortRequest();
};
request.raw.once('aborted', abortRequest);
reply.raw.once('close', abortClosedReply);
if (request.raw.aborted) abortRequest();
try {
return {
ok: true,
...(await tasks.listOperationsDashboard(contextFor(session, request), {
from: query.from,
to: query.to,
actorUserId: query.actor_user_id,
businessRouteId: query.business_route_id,
status: query.status,
search: query.search,
limit: query.limit,
offset: query.offset,
signal: controller.signal
}))
};
} finally {
request.raw.off('aborted', abortRequest);
reply.raw.off('close', abortClosedReply);
}
});
app.get('/api/operations-dashboard/tasks/:taskId', async (request, reply) => {

390
control-plane/src/task-service.ts
View File

@@ -1895,6 +1895,9 @@ const OPERATIONS_DASHBOARD_FALLBACK_TYPES = [
{ key: 'dashboard_order_update', label: '订单修改', action: 'order_update' }
] as const;
const OPERATIONS_DASHBOARD_QUERY_BUDGET_MS = 9_000;
const OPERATIONS_DASHBOARD_STATEMENT_TIMEOUT_MS = 5_000;
function operationsDashboardFallbackType(value: unknown, fallbackText: unknown) {
const operation = operationsDashboardOperationObject(value, fallbackText);
const action = text(operation.action).trim();
@@ -1909,13 +1912,6 @@ function operationsDashboardTypeById(value: unknown) {
return OPERATIONS_DASHBOARD_FALLBACK_TYPES.find((candidate) => candidate.key === key) || null;
}
function operationsDashboardNeedsInstructionForType(routeId: unknown, operation: unknown): boolean {
if (businessRouteById(routeId)) return false;
if (operationsDashboardRouteFromOperation(operation, '')) return false;
const fallback = operationsDashboardFallbackType(operation, '');
return !fallback || fallback.action === 'arrangement_hotel' || fallback.action === 'passenger_list_import';
}
function operationsDashboardBusiness(
routeId: unknown,
instruction?: unknown,
@@ -4663,9 +4659,30 @@ export class TaskService {
search?: string;
limit?: number;
offset?: number;
signal?: AbortSignal;
} = {}
): Promise<PublicOperationsDashboardPage> {
this.requireOperationsDashboardAccess(context);
const requestStartedAt = Date.now();
const queryDeadlineAt = requestStartedAt + OPERATIONS_DASHBOARD_QUERY_BUDGET_MS;
const timeoutError = () => new TaskError(
'operations_dashboard_query_timeout',
'看板查询耗时过长，请缩短日期范围或增加筛选条件后重试。',
503
);
const logTimeout = (errorCode: string) => this.logger.warn({
diagnostic_event: 'operations_dashboard.timeout',
diagnostic_stage: 'operations_dashboard',
request_id: context.requestId,
duration_ms: Date.now() - requestStartedAt,
error_code: errorCode
}, 'operations dashboard timed out');
const assertQueryAvailable = () => {
if (options.signal?.aborted) {
throw new TaskError('operations_dashboard_request_aborted', '看板请求已取消。', 499);
}
if (Date.now() >= queryDeadlineAt) throw timeoutError();
};
const now = new Date();
const to = options.to ? new Date(options.to) : now;
const from = options.from
@@ -4694,25 +4711,23 @@ export class TaskService {
if (businessRouteId && businessRouteId !== 'unclassified' && !operationsDashboardTypeById(businessRouteId)) {
throw new TaskError('operations_dashboard_business_invalid', '看板业务类型无效。', 400);
}
if (businessRouteId === 'unclassified') {
where += ' AND t.business_route_id IS NULL';
} else if (businessRouteId) {
params.push(businessRouteId);
where += ` AND (t.business_route_id = $${params.length} OR t.business_route_id IS NULL)`;
}
const statusCondition = operationsStatusSql(status);
if (statusCondition) where += ` AND ${statusCondition}`;
const search = text(options.search).trim().toLowerCase();
const boundedLimit = Math.max(1, Math.min(100, Math.trunc(options.limit || 50)));
const boundedLimit = Math.max(1, Math.min(100, Math.trunc(options.limit || 20)));
const boundedOffset = Math.max(0, Math.min(1_000_000, Math.trunc(options.offset || 0)));
const pool = getPool(this.config);
const businessOptions = [
...BUSINESS_ROUTES.map((route) => ({ key: route.routeId, label: route.directive })),
...OPERATIONS_DASHBOARD_FALLBACK_TYPES.map(({ key, label }) => ({ key, label })),
{ key: 'unclassified', label: '其他任务' }
];
const actorQuery = () => pool.query(
`SELECT id, username, role, is_active
FROM users
WHERE organization_id = $1
ORDER BY is_active DESC, username ASC`,
[context.organizationId]
);
const taskBaseSelect = `t.id, t.task_id, t.status, t.business_route_id, t.archived_at,
t.created_at, t.updated_at, t.operation,
creator.id AS actor_id, creator.username AS actor_username,
@@ -4727,154 +4742,217 @@ export class TaskService {
AND input_message.task_id = t.id
AND input_message.role = 'user'
AND input_message.input_source IS DISTINCT FROM 'reparse') AS input_turn_count`;
const loadDetailedRows = async (taskRowIds: string[]) => {
if (!taskRowIds.length) return [] as Record<string, unknown>[];
const detailResult = await pool.query(
`SELECT ${taskDetailSelect}
FROM tasks t
LEFT JOIN users creator ON creator.id = t.created_by
WHERE t.organization_id = $1
AND t.id = ANY($2::uuid[])`,
[context.organizationId, taskRowIds]
);
return detailResult.rows as Record<string, unknown>[];
};
const taskSearchSelect = `${taskBaseSelect},
t.original_text_ciphertext, t.message, t.success_receipt, t.success_receipt_at,
t.error_summary, t.error_summary_at,
CASE
WHEN t.status = 'completed'
AND (t.success_receipt IS NULL OR t.success_receipt = '{}'::jsonb)
THEN t.execution_result
ELSE NULL
END AS execution_result,
CASE WHEN t.status = 'awaiting_user_input' THEN t.parse_response ELSE NULL END AS parse_response`;
const taskCandidateSelect = `${taskBaseSelect},
CASE WHEN t.business_route_id IS NULL THEN t.original_text_ciphertext ELSE NULL END AS original_text_ciphertext`;
const [candidateResult, actorResult] = await Promise.all([
pool.query(
`SELECT ${taskBaseSelect}
FROM tasks t
LEFT JOIN users creator ON creator.id = t.created_by
WHERE ${where}
ORDER BY t.created_at DESC, t.id DESC`,
params
),
actorQuery()
]);
if (search && candidateResult.rows.length > 2_000) {
throw new TaskError(
'operations_dashboard_search_scope_too_large',
'关键词查询范围超过 2000 项任务，请先缩短日期或选择人员、任务类型后再查询。',
400
);
}
try {
assertQueryAvailable();
return await withTransaction(this.config, async (client) => {
assertQueryAvailable();
await client.query('SET TRANSACTION READ ONLY');
await client.query(`SET LOCAL statement_timeout = '${OPERATIONS_DASHBOARD_STATEMENT_TIMEOUT_MS}ms'`);
const candidateRows = candidateResult.rows as Record<string, unknown>[];
const searchValuesByTask = new Map<string, string[]>();
let projectionRows = candidateRows;
if (search && candidateRows.length) {
const taskRowIds = candidateRows.map((row) => text(row.id)).filter(Boolean);
const [detailedRows, messageResult] = await Promise.all([
loadDetailedRows(taskRowIds),
pool.query(
`SELECT m.task_id, m.content_ciphertext, actor.username AS actor_username
FROM agent_session_messages m
LEFT JOIN users actor ON actor.id = m.actor_user_id
WHERE m.organization_id = $1
AND m.task_id = ANY($2::uuid[])
AND m.role = 'user'
AND m.input_source IS DISTINCT FROM 'reparse'`,
[context.organizationId, taskRowIds]
)
]);
const detailedById = new Map(detailedRows.map((row) => [text(row.id), row]));
projectionRows = candidateRows.map((row) => detailedById.get(text(row.id)) || row);
for (const row of messageResult.rows as Record<string, unknown>[]) {
const key = text(row.task_id);
const values = searchValuesByTask.get(key) || [];
values.push(
text(row.actor_username),
operationsDashboardLeadershipInstructionText(
decryptText(this.config, row.content_ciphertext as string | null | undefined)
)
const dashboardQuery = async (
queryStage: string,
sql: string,
queryParams: unknown[] = []
) => {
assertQueryAvailable();
const startedAt = Date.now();
const result = await client.query(sql, queryParams);
this.logger.info({
diagnostic_event: 'operations_dashboard.query.completed',
diagnostic_stage: 'operations_dashboard_query',
request_id: context.requestId,
query_stage: queryStage,
duration_ms: Date.now() - startedAt,
row_count: result.rowCount ?? 0
}, 'operations dashboard query completed');
assertQueryAvailable();
return result;
};
const loadDetailedRows = async (taskRowIds: string[]) => {
if (!taskRowIds.length) return [] as Record<string, unknown>[];
const detailResult = await dashboardQuery(
'page_details',
`SELECT ${taskDetailSelect}
FROM tasks t
LEFT JOIN users creator ON creator.id = t.created_by
WHERE t.organization_id = $1
AND t.id = ANY($2::uuid[])`,
[context.organizationId, taskRowIds]
);
return detailResult.rows as Record<string, unknown>[];
};
const candidateResult = await dashboardQuery(
'candidates',
`SELECT ${search ? taskSearchSelect : taskCandidateSelect}
FROM tasks t
LEFT JOIN users creator ON creator.id = t.created_by
WHERE ${where}
ORDER BY t.created_at DESC, t.id DESC`,
params
);
searchValuesByTask.set(key, values);
}
} else if (candidateRows.length) {
const classificationIds = candidateRows
.filter((row) => operationsDashboardNeedsInstructionForType(row.business_route_id, row.operation))
.map((row) => text(row.id))
.filter(Boolean);
if (classificationIds.length) {
const classificationResult = await pool.query(
`SELECT id, original_text_ciphertext
FROM tasks
if (search && candidateResult.rows.length > 2_000) {
throw new TaskError(
'operations_dashboard_search_scope_too_large',
'关键词查询范围超过 2000 项任务，请先缩短日期或选择人员、任务类型后再查询。',
400
);
}
const actorResult = await dashboardQuery(
'actors',
`SELECT id, username, role, is_active
FROM users
WHERE organization_id = $1
AND id = ANY($2::uuid[])`,
[context.organizationId, classificationIds]
ORDER BY is_active DESC, username ASC`,
[context.organizationId]
);
const instructionsById = new Map(
(classificationResult.rows as Record<string, unknown>[])
.map((row) => [text(row.id), row.original_text_ciphertext])
);
projectionRows = candidateRows.map((row) => ({
...row,
original_text_ciphertext: instructionsById.get(text(row.id))
}));
}
}
const candidateRows = candidateResult.rows as Record<string, unknown>[];
const projectedCandidates = candidateRows.flatMap((row) => {
const projection = operationsDashboardTaskProjection(this.config, row);
const routeMatches = !businessRouteId
|| (businessRouteId === 'unclassified'
? !projection.task.business_route_id
: projection.task.business_route_id === businessRouteId);
if (!routeMatches) return [];
const directSearchMatch = !search || operationsDashboardSearchMatches(search, [
projection.task.creator.username,
projection.task.business_route_label,
projection.instruction,
projection.result
]);
return [{ row, task: projection.task, directSearchMatch }];
});
assertQueryAvailable();
const searchValuesByTask = new Map<string, string[]>();
if (search && projectedCandidates.length) {
const taskRowIds = projectedCandidates
.filter(({ directSearchMatch }) => !directSearchMatch)
.map(({ row }) => text(row.id))
.filter(Boolean);
if (taskRowIds.length) {
const messageResult = await dashboardQuery(
'search_messages',
`SELECT m.task_id, m.content_ciphertext, actor.username AS actor_username
FROM agent_session_messages m
LEFT JOIN users actor ON actor.id = m.actor_user_id
WHERE m.organization_id = $1
AND m.task_id = ANY($2::uuid[])
AND m.role = 'user'
AND m.input_source IS DISTINCT FROM 'reparse'`,
[context.organizationId, taskRowIds]
);
for (const row of messageResult.rows as Record<string, unknown>[]) {
const key = text(row.task_id);
const values = searchValuesByTask.get(key) || [];
values.push(
text(row.actor_username),
operationsDashboardLeadershipInstructionText(
decryptText(this.config, row.content_ciphertext as string | null | undefined)
)
);
searchValuesByTask.set(key, values);
}
}
}
const matched = projectionRows.flatMap((row) => {
const projection = operationsDashboardTaskProjection(this.config, row);
const routeMatches = !businessRouteId
|| (businessRouteId === 'unclassified'
? !projection.task.business_route_id
: projection.task.business_route_id === businessRouteId);
if (!routeMatches) return [];
if (search && !operationsDashboardSearchMatches(search, [
projection.task.creator.username,
projection.task.business_route_label,
projection.instruction,
projection.result,
...(searchValuesByTask.get(text(row.id)) || [])
])) return [];
return [{ row, task: projection.task }];
});
const matchedTasks = matched.map(({ task }) => task);
const actors = (actorResult.rows as Record<string, unknown>[]).map((row) => ({
id: text(row.id),
username: text(row.username),
role: taskRole(row.role),
is_active: databaseBoolean(row.is_active)
} satisfies PublicOperationsDashboardActor));
const summary = operationsDashboardSummaryFromTasks(matchedTasks);
const days = operationsDashboardBreakdownsFromTasks(
matchedTasks,
(task) => ({ key: task.business_day, label: task.business_day }),
(left, right) => right.key.localeCompare(left.key)
);
const businesses = operationsDashboardBreakdownsFromTasks(
matchedTasks,
(task) => ({
key: task.business_route_id || 'unclassified',
label: task.business_route_label
}),
(left, right) => right.total - left.total || left.label.localeCompare(right.label, 'zh-CN')
);
const pageMatches = matched.slice(boundedOffset, boundedOffset + boundedLimit);
let pageTasks = pageMatches.map(({ task }) => task);
if (!search && pageMatches.length) {
const detailedRows = await loadDetailedRows(pageMatches.map(({ row }) => text(row.id)));
const detailedById = new Map(detailedRows.map((row) => [text(row.id), row]));
pageTasks = pageMatches.map(({ row, task }) => {
const detailed = detailedById.get(text(row.id));
return detailed ? operationsDashboardTaskProjection(this.config, detailed).task : task;
assertQueryAvailable();
const matched = projectedCandidates.filter(({ row, directSearchMatch }) => (
directSearchMatch
|| operationsDashboardSearchMatches(search, searchValuesByTask.get(text(row.id)) || [])
));
assertQueryAvailable();
const matchedTasks = matched.map(({ task }) => task);
const actors = (actorResult.rows as Record<string, unknown>[]).map((row) => ({
id: text(row.id),
username: text(row.username),
role: taskRole(row.role),
is_active: databaseBoolean(row.is_active)
} satisfies PublicOperationsDashboardActor));
const summary = operationsDashboardSummaryFromTasks(matchedTasks);
const days = operationsDashboardBreakdownsFromTasks(
matchedTasks,
(task) => ({ key: task.business_day, label: task.business_day }),
(left, right) => right.key.localeCompare(left.key)
);
const businesses = operationsDashboardBreakdownsFromTasks(
matchedTasks,
(task) => ({
key: task.business_route_id || 'unclassified',
label: task.business_route_label
}),
(left, right) => right.total - left.total || left.label.localeCompare(right.label, 'zh-CN')
);
const pageMatches = matched.slice(boundedOffset, boundedOffset + boundedLimit);
let pageTasks = pageMatches.map(({ task }) => task);
if (pageMatches.length) {
const detailedRows = await loadDetailedRows(pageMatches.map(({ row }) => text(row.id)));
const detailedById = new Map(detailedRows.map((row) => [text(row.id), row]));
pageTasks = pageMatches.map(({ row, task }) => {
const detailed = detailedById.get(text(row.id));
return detailed ? operationsDashboardTaskProjection(this.config, detailed).task : task;
});
}
assertQueryAvailable();
this.logger.info({
diagnostic_event: 'operations_dashboard.completed',
diagnostic_stage: 'operations_dashboard',
request_id: context.requestId,
duration_ms: Date.now() - requestStartedAt,
candidate_count: candidateRows.length,
matched_count: matchedTasks.length,
page_count: pageTasks.length,
keyword_filter: Boolean(search),
business_filter: Boolean(businessRouteId),
actor_filter: Boolean(actorUserId),
status_filter: status !== 'all'
}, 'operations dashboard completed');
return {
summary,
users: operationsDashboardUsersFromTasks(matchedTasks),
actors,
days,
businesses,
business_options: businessOptions,
tasks: pageTasks,
range: { from: from.toISOString(), to: to.toISOString() },
total: matchedTasks.length,
offset: boundedOffset,
limit: boundedLimit,
has_more: boundedOffset + pageTasks.length < matchedTasks.length
};
});
} catch (error) {
if (error instanceof TaskError) {
if (error.code === 'operations_dashboard_query_timeout') logTimeout(error.code);
throw error;
}
const databaseCode = error && typeof error === 'object' && 'code' in error
? text((error as { code?: unknown }).code)
: '';
const errorMessage = error instanceof Error ? error.message : text(error);
if (options.signal?.aborted) {
throw new TaskError('operations_dashboard_request_aborted', '看板请求已取消。', 499);
}
if (databaseCode === '57014' || /timeout exceeded when trying to connect|query read timeout/i.test(errorMessage)) {
logTimeout(databaseCode || 'database_timeout');
throw timeoutError();
}
throw error;
}
return {
summary,
users: operationsDashboardUsersFromTasks(matchedTasks),
actors,
days,
businesses,
business_options: businessOptions,
tasks: pageTasks,
range: { from: from.toISOString(), to: to.toISOString() },
total: matchedTasks.length,
offset: boundedOffset,
limit: boundedLimit,
has_more: boundedOffset + pageTasks.length < matchedTasks.length
};
}
async getOperationsDashboardTask(

2
control-plane/test/account-authorization.test.ts
View File

@@ -135,7 +135,6 @@ test('operations dashboard is leadership-gated, business-facing, searchable, and
assert.match(tasks, /operations_dashboard_search_scope_too_large/);
assert.match(tasks, /function operationsDashboardRouteFromInstruction/);
assert.match(tasks, /function operationsDashboardRouteFromOperation/);
assert.match(tasks, /function operationsDashboardNeedsInstructionForType/);
assert.match(tasks, /if \(search && candidateResult\.rows\.length > 2_000\)/);
assert.match(tasks, /OPERATIONS_DASHBOARD_FALLBACK_TYPES/);
assert.match(tasks, /dashboard_order_delete/);
@@ -173,6 +172,7 @@ test('operations dashboard is leadership-gated, business-facing, searchable, and
assert.match(server, /app\.get\('\/api\/operations-dashboard'[\s\S]+requireLeadershipSession\(request\)/);
assert.match(server, /business_route_id: z\.string\(\)\.trim\(\)\.max\(120\)\.optional\(\)/);
assert.match(server, /businessRouteId: query\.business_route_id/);
assert.match(server, /signal: controller\.signal/);
assert.match(server, /app\.get\('\/api\/operations-dashboard\/tasks\/:taskId'[\s\S]+requireLeadershipSession\(request\)/);
assert.match(server, /read_only: true/);
assert.doesNotMatch(server, /app\.(?:post|put|patch|delete)\('\/api\/operations-dashboard/);

2
control-plane/test/control-plane.test.ts
View File

@@ -1226,7 +1226,7 @@ test('operator page has a login gate and uses the durable task API', async () =>
assert.match(index, /id="loginPanel"/);
assert.match(index, /id="workbench"[^>]*hidden/);
assert.match(index, /styles\.css\?v=20260902-dashboard-mobile-share-1/);
assert.match(index, /app\.js\?v=20260902-dashboard-mobile-share-1/);
assert.match(index, /app\.js\?v=20260902-dashboard-filter-fix-2/);
assert.match(index, /id="statusDetailsPopover"/);
assert.match(index, /id="statusDetailsRefresh"/);
assert.match(app, /apiRequest\(`\/api\/tasks\?\$\{params\.toString\(\)\}`/);