refactor(tool-session-query): pass projection handles to operations

This commit is contained in:
_Kerman committed 2026-10-12 13:08:18 +08:00
1 parent 99f3660120
commit 5b401b27a5
5 files changed
+215 -244

No files matched your search

+1 -1
View File
@@ -3921,7 +3921,7 @@ export interface Config {
## `@deepseek-ai/dsh-tool-session-query`
- `inject`: `tools` · `systemPrompt` · `sessionQuery` · `sessionProjections`
- `source`: [`packages/experimental/tool-session-query/src/index.ts:26`](../packages/experimental/tool-session-query/src/index.ts)
- `source`: [`packages/experimental/tool-session-query/src/index.ts:27`](../packages/experimental/tool-session-query/src/index.ts)
```ts config-catalog
/** Deployment-owned search count and timeout bounds. */
+1 -1
View File
@@ -3923,7 +3923,7 @@ export interface Config {
## `@deepseek-ai/dsh-tool-session-query`
- `inject`: `tools` · `systemPrompt` · `sessionQuery` · `sessionProjections`
- `source`: [`packages/experimental/tool-session-query/src/index.ts:26`](../packages/experimental/tool-session-query/src/index.ts)
- `source`: [`packages/experimental/tool-session-query/src/index.ts:27`](../packages/experimental/tool-session-query/src/index.ts)
```ts config-catalog
/** Deployment-owned search count and timeout bounds. */
@@ -5,11 +5,12 @@
*/
import type { Context } from '@deepseek-ai/cordis'
import { turnBoundaryProjectionDefinition } from '@deepseek-ai/dsh-agent'
import z from '@deepseek-ai/schemastery'
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
import { defineTool } from '@deepseek-ai/dsh-tools'
import { toolInput } from './input.ts'
import { SessionQueryOperations } from './operations.ts'
import { operations } from './operations.ts'
import { presentation } from './presentation.ts'
export const name = 'tool-session-query'
@@ -58,8 +59,7 @@ const PROMPT_TEXT =
* @returns when the tools and their projection reader are ready.
*/
export async function apply(ctx: Context, config: Config): Promise<void> {
const operations = new SessionQueryOperations(ctx)
await operations.initialize()
const turnBoundary = await ctx.sessionProjections.register(turnBoundaryProjectionDefinition)
const resolved = resolveConfig(config)
ctx.systemPrompt.section({
name: 'tool:session-query',
@@ -73,7 +73,7 @@ export async function apply(ctx: Context, config: Config): Promise<void> {
parameters: toolInput.sessionSearchParameters,
output: TEXT_OUTPUT,
timeoutMs: resolved.searchTimeoutMs,
execute: (args, exec) => operations.executeSessionSearch(args, exec, resolved.maxSearchResults),
execute: (args, exec) => operations.executeSessionSearch(ctx, turnBoundary, args, exec, resolved.maxSearchResults),
presentCall: presentation.presentSessionSearchCall,
}))
@@ -83,7 +83,7 @@ export async function apply(ctx: Context, config: Config): Promise<void> {
parameters: toolInput.eventSearchParameters,
output: TEXT_OUTPUT,
timeoutMs: resolved.searchTimeoutMs,
execute: (args, exec) => operations.executeEventSearch(args, exec, resolved.maxSearchResults),
execute: (args, exec) => operations.executeEventSearch(ctx, turnBoundary, args, exec, resolved.maxSearchResults),
presentCall: presentation.presentEventSearchCall,
}))
@@ -93,7 +93,7 @@ export async function apply(ctx: Context, config: Config): Promise<void> {
parameters: toolInput.targetSessionParameter,
output: TEXT_OUTPUT,
isConcurrencySafe: () => true,
execute: (args, exec) => operations.executeSessionTrace(args, exec),
execute: (args, exec) => operations.executeSessionTrace(ctx, turnBoundary, args, exec),
presentCall: presentation.presentSessionTraceCall,
}))
@@ -106,7 +106,7 @@ export async function apply(ctx: Context, config: Config): Promise<void> {
},
output: TEXT_OUTPUT,
isConcurrencySafe: () => true,
execute: (args, exec) => operations.executeEventTrace(args, exec),
execute: (args, exec) => operations.executeEventTrace(ctx, turnBoundary, args, exec),
presentCall: args => presentation.presentEventTargetCall('Trace event', args),
}))
@@ -121,7 +121,7 @@ export async function apply(ctx: Context, config: Config): Promise<void> {
},
output: TEXT_OUTPUT,
isConcurrencySafe: () => true,
execute: (args, exec) => operations.executeEventRead(args, exec),
execute: (args, exec) => operations.executeEventRead(ctx, turnBoundary, args, exec),
presentCall: args => presentation.presentEventTargetCall('Read event', args),
}))
}
@@ -4,9 +4,9 @@
* @module @deepseek-ai/dsh-tool-session-query/operations
*/
import { turnBoundaryProjectionDefinition, type TurnBoundaryProjection } from '@deepseek-ai/dsh-agent'
import type { ProjectionHandle } from '@deepseek-ai/dsh-session-projection'
import type { Context } from '@deepseek-ai/cordis'
import type { TurnBoundaryProjection } from '@deepseek-ai/dsh-agent'
import type { ProjectionHandle } from '@deepseek-ai/dsh-session-projection'
import { HarnessError } from '@deepseek-ai/dsh-llm'
import { SessionSeq } from '@deepseek-ai/dsh-session'
import type { SessionId } from '@deepseek-ai/dsh-session'
@@ -54,249 +54,192 @@ interface SearchCollection<T> {
readonly capped: boolean
}
/** Session-history tool operations over the caller’s current turn state. */
export class SessionQueryOperations {
private turnBoundary!: ProjectionHandle<TurnBoundaryProjection>
constructor(private readonly ctx: Context) {}
/**
* Acquire projection readers before exposing dependent operations.
* @returns when all required registrations are ready.
*/
async initialize(): Promise<void> {
this.turnBoundary = await this.ctx.sessionProjections.register(turnBoundaryProjectionDefinition)
async function executeSessionSearch(
ctx: Context,
turnBoundary: ProjectionHandle<TurnBoundaryProjection>,
args: SessionSearchArgs,
exec: ToolRunContext,
maxResults: number,
): Promise<string> {
const caller = workspaceAccess.callerOf(exec, turnBoundary)
const cwd = caller.header.cwd
if (cwd === undefined) {
throw new HarnessError(
'cross-session search is unavailable because the caller session has no workspace',
'SESSION_QUERY_TOOL_UNAUTHORIZED',
)
}
private callerOf(exec: ToolRunContext) {
const agent = exec.agent
if (agent === undefined) {
throw new HarnessError(
'session query tools require an agent-bound caller',
'SESSION_QUERY_TOOL_MISSING_AGENT',
)
}
return {
id: agent.session.id,
header: agent.session.header,
boundary: this.turnBoundary.stateOf(agent.session),
}
const query = toolInput.normalizeQuery(args.query)
const sessionFilters = toolInput.buildSessionFilters(args)
const eventFilters = toolInput.buildEventFilters({
seqFrom: args.event_seq_from,
seqTo: args.event_seq_to,
timeFrom: args.event_time_from,
timeTo: args.event_time_to,
eventTypes: args.event_types,
surfaces: args.event_surfaces,
})
const requestedParentIds = toolInput.materializeParentSessionIds(args.parent_session_ids)
if (requestedParentIds !== undefined || args.include_root_sessions === true) {
const authorizedParentIds = requestedParentIds === undefined
? new Set<SessionId>()
: await workspaceAccess.authorizeSessionIds(ctx, caller, requestedParentIds, exec.signal)
const parentValues: Array<SessionId | null> = requestedParentIds
?.filter(id => authorizedParentIds.has(id)) ?? []
if (args.include_root_sessions === true) parentValues.push(null)
if (parentValues.length === 0) return presentation.formatEmptySessionSearch()
sessionFilters.push({ kind: 'parent', values: parentValues })
}
sessionFilters.push({ kind: 'cwd', values: [cwd] })
const collected = await collectPages(
maxResults,
exec.signal,
cursor => serviceBoundary.call(ctx, exec.signal, 'session search', () =>
ctx.sessionQuery.searchSessions({
query,
sessionFilters,
eventFilters,
...cursor === undefined ? {} : { cursor },
}, { signal: exec.signal })),
hit => hit.header.id !== caller.id && workspaceAccess.recordAuthorized(hit, caller),
)
/**
* Search other Sessions within the caller’s authorized workspace.
* @param args - Session and event filters.
* @param exec - Agent-bound caller and cancellation signal.
* @param maxResults - Maximum number of authorized matches to return.
* @returns Formatted authorized Session matches and truncation status.
*/
async executeSessionSearch(
args: SessionSearchArgs,
exec: ToolRunContext,
maxResults: number,
): Promise<string> {
const ctx = this.ctx
const caller = this.callerOf(exec)
const cwd = caller.header.cwd
if (cwd === undefined) {
throw new HarnessError(
'cross-session search is unavailable because the caller session has no workspace',
'SESSION_QUERY_TOOL_UNAUTHORIZED',
)
}
const query = toolInput.normalizeQuery(args.query)
const sessionFilters = toolInput.buildSessionFilters(args)
const eventFilters = toolInput.buildEventFilters({
seqFrom: args.event_seq_from,
seqTo: args.event_seq_to,
timeFrom: args.event_time_from,
timeTo: args.event_time_to,
eventTypes: args.event_types,
surfaces: args.event_surfaces,
})
const requestedParentIds = toolInput.materializeParentSessionIds(args.parent_session_ids)
if (requestedParentIds !== undefined || args.include_root_sessions === true) {
const authorizedParentIds = requestedParentIds === undefined
? new Set<SessionId>()
: await workspaceAccess.authorizeSessionIds(ctx, caller, requestedParentIds, exec.signal)
const parentValues: Array<SessionId | null> = requestedParentIds
?.filter(id => authorizedParentIds.has(id)) ?? []
if (args.include_root_sessions === true) parentValues.push(null)
if (parentValues.length === 0) return presentation.formatEmptySessionSearch()
sessionFilters.push({ kind: 'parent', values: parentValues })
}
sessionFilters.push({ kind: 'cwd', values: [cwd] })
const collected = await collectPages(
maxResults,
exec.signal,
cursor => serviceBoundary.call(ctx, exec.signal, 'session search', () =>
ctx.sessionQuery.searchSessions({
const parentIds = collected.items
.map(hit => hit.header.parentSession)
.filter((id): id is SessionId => id !== undefined)
const authorizedParents = await workspaceAccess.authorizeSessionIds(ctx, caller, parentIds, exec.signal)
const titles = await workspaceAccess.readTitles(
ctx,
caller,
collected.items.map(hit => hit.header.id),
exec.signal,
)
return presentation.formatSessionSearch(collected, titles, authorizedParents)
}
async function executeEventSearch(
ctx: Context,
turnBoundary: ProjectionHandle<TurnBoundaryProjection>,
args: EventSearchArgs,
exec: ToolRunContext,
maxResults: number,
): Promise<string> {
const caller = workspaceAccess.callerOf(exec, turnBoundary)
const sessionId = workspaceAccess.targetId(args, caller)
await workspaceAccess.authorizeTarget(ctx, caller, sessionId, exec.signal)
const query = toolInput.normalizeQuery(args.query)
const range = toolInput.sequenceRange(args.seq_from, args.seq_to)
if (sessionId === caller.id) {
const stepStartSeq = caller.boundary.lastStepStartSeq
// `null` (no step started yet) caps the range to a degenerate `to` the
// filter validation rejects — the loop never runs tools outside a step.
range.to = Math.min(range.to ?? Number.MAX_SAFE_INTEGER, (stepStartSeq ?? 0) - 1)
}
const title = await workspaceAccess.readTitle(ctx, caller, sessionId, exec.signal)
if (range.from !== undefined && range.to !== undefined && range.from > range.to) {
return presentation.formatEventSearch(sessionId, title, { items: [], capped: false })
}
const filters = toolInput.buildEventFilters({
seqFrom: range.from,
seqTo: range.to,
timeFrom: args.time_from,
timeTo: args.time_to,
eventTypes: args.event_types,
surfaces: args.surfaces,
})
const collected = await collectPages(
maxResults,
exec.signal,
async (cursor): Promise<SessionEventSearchPage> => {
const page = await serviceBoundary.call(ctx, exec.signal, 'event search', () =>
ctx.sessionQuery.searchEvents({
sessionId,
query,
sessionFilters,
eventFilters,
filters,
...cursor === undefined ? {} : { cursor },
}, { signal: exec.signal })),
hit => hit.header.id !== caller.id && workspaceAccess.recordAuthorized(hit, caller),
)
}, { signal: exec.signal }))
workspaceAccess.assertObservedTargetAuthorized(caller, sessionId, page.session)
return page
},
() => true,
)
return presentation.formatEventSearch(sessionId, title, collected)
}
const parentIds = collected.items
.map(hit => hit.header.parentSession)
.filter((id): id is SessionId => id !== undefined)
const authorizedParents = await workspaceAccess.authorizeSessionIds(ctx, caller, parentIds, exec.signal)
const titles = await workspaceAccess.readTitles(
ctx,
caller,
collected.items.map(hit => hit.header.id),
exec.signal,
)
return presentation.formatSessionSearch(collected, titles, authorizedParents)
}
async function executeSessionTrace(
ctx: Context,
turnBoundary: ProjectionHandle<TurnBoundaryProjection>,
args: SessionTargetArgs,
exec: ToolRunContext,
): Promise<string> {
const caller = workspaceAccess.callerOf(exec, turnBoundary)
const sessionId = workspaceAccess.targetId(args, caller)
await workspaceAccess.authorizeTarget(ctx, caller, sessionId, exec.signal)
const trace = await serviceBoundary.call(ctx, exec.signal, 'session lineage trace', () =>
ctx.sessionQuery.traceSession(sessionId, exec.signal))
workspaceAccess.assertObservedTargetAuthorized(caller, sessionId, trace.target.header)
/**
* Search authorized Session events, excluding the caller’s current step.
* @param args - Target Session and event filters.
* @param exec - Agent-bound caller and cancellation signal.
* @param maxResults - Maximum number of authorized matches to return.
* @returns Formatted matching events and truncation status.
*/
async executeEventSearch(
args: EventSearchArgs,
exec: ToolRunContext,
maxResults: number,
): Promise<string> {
const ctx = this.ctx
const caller = this.callerOf(exec)
const sessionId = workspaceAccess.targetId(args, caller)
await workspaceAccess.authorizeTarget(ctx, caller, sessionId, exec.signal)
const query = toolInput.normalizeQuery(args.query)
const range = toolInput.sequenceRange(args.seq_from, args.seq_to)
if (sessionId === caller.id) {
const stepStartSeq = caller.boundary.lastStepStartSeq
// `null` (no step started yet) caps the range to a degenerate `to` the
// filter validation rejects — the loop never runs tools outside a step.
range.to = Math.min(range.to ?? Number.MAX_SAFE_INTEGER, (stepStartSeq ?? 0) - 1)
const ancestors: SessionRecord[] = []
let ancestorBoundary = false
for (const ancestor of trace.ancestors) {
if (!workspaceAccess.recordAuthorized(ancestor, caller)) {
ancestorBoundary = true
break
}
const title = await workspaceAccess.readTitle(ctx, caller, sessionId, exec.signal)
if (range.from !== undefined && range.to !== undefined && range.from > range.to) {
return presentation.formatEventSearch(sessionId, title, { items: [], capped: false })
}
const filters = toolInput.buildEventFilters({
seqFrom: range.from,
seqTo: range.to,
timeFrom: args.time_from,
timeTo: args.time_to,
eventTypes: args.event_types,
surfaces: args.surfaces,
})
const collected = await collectPages(
maxResults,
exec.signal,
async (cursor): Promise<SessionEventSearchPage> => {
const page = await serviceBoundary.call(ctx, exec.signal, 'event search', () =>
ctx.sessionQuery.searchEvents({
sessionId,
query,
filters,
...cursor === undefined ? {} : { cursor },
}, { signal: exec.signal }))
workspaceAccess.assertObservedTargetAuthorized(caller, sessionId, page.session)
return page
},
() => true,
)
return presentation.formatEventSearch(sessionId, title, collected)
ancestors.push(ancestor)
}
if (ancestors.length === trace.ancestors.length && !trace.complete) ancestorBoundary = true
const descendants = workspaceAccess.authorizeDescendants(trace.descendants, caller)
const visibleIds = [
trace.target.header.id,
...ancestors.map(record => record.header.id),
...workspaceAccess.descendantIds(descendants),
]
const titles = await workspaceAccess.readTitles(ctx, caller, visibleIds, exec.signal)
return presentation.formatSessionTrace(trace, ancestors, ancestorBoundary, descendants, titles)
}
/**
* Trace Session lineage within the caller’s workspace permissions.
* @param args - Optional target Session id; defaults to the caller.
* @param exec - Agent-bound caller and cancellation signal.
* @returns Formatted authorized ancestors and descendants.
*/
async executeSessionTrace(
args: SessionTargetArgs,
exec: ToolRunContext,
): Promise<string> {
const ctx = this.ctx
const caller = this.callerOf(exec)
const sessionId = workspaceAccess.targetId(args, caller)
await workspaceAccess.authorizeTarget(ctx, caller, sessionId, exec.signal)
const trace = await serviceBoundary.call(ctx, exec.signal, 'session lineage trace', () =>
ctx.sessionQuery.traceSession(sessionId, exec.signal))
workspaceAccess.assertObservedTargetAuthorized(caller, sessionId, trace.target.header)
async function executeEventTrace(
ctx: Context,
turnBoundary: ProjectionHandle<TurnBoundaryProjection>,
args: EventTargetArgs,
exec: ToolRunContext,
): Promise<string> {
toolInput.assertNonNegativeSafeInteger('seq', args.seq)
const seq = SessionSeq(args.seq)
const caller = workspaceAccess.callerOf(exec, turnBoundary)
const sessionId = workspaceAccess.targetId(args, caller)
await workspaceAccess.authorizeTarget(ctx, caller, sessionId, exec.signal)
const trace = await serviceBoundary.call(ctx, exec.signal, 'event trace', () =>
ctx.sessionQuery.traceEvent({ sessionId, seq }, exec.signal))
workspaceAccess.assertObservedTargetAuthorized(caller, sessionId, trace.session)
const title = await workspaceAccess.readTitle(ctx, caller, sessionId, exec.signal)
return presentation.formatEventTrace(sessionId, title, trace)
}
const ancestors: SessionRecord[] = []
let ancestorBoundary = false
for (const ancestor of trace.ancestors) {
if (!workspaceAccess.recordAuthorized(ancestor, caller)) {
ancestorBoundary = true
break
}
ancestors.push(ancestor)
}
if (ancestors.length === trace.ancestors.length && !trace.complete) ancestorBoundary = true
const descendants = workspaceAccess.authorizeDescendants(trace.descendants, caller)
const visibleIds = [
trace.target.header.id,
...ancestors.map(record => record.header.id),
...workspaceAccess.descendantIds(descendants),
]
const titles = await workspaceAccess.readTitles(ctx, caller, visibleIds, exec.signal)
return presentation.formatSessionTrace(trace, ancestors, ancestorBoundary, descendants, titles)
}
/**
* Trace an event after authorizing its owning Session.
* @param args - Target Session and event sequence.
* @param exec - Agent-bound caller and cancellation signal.
* @returns Formatted event lineage.
*/
async executeEventTrace(
args: EventTargetArgs,
exec: ToolRunContext,
): Promise<string> {
const ctx = this.ctx
toolInput.assertNonNegativeSafeInteger('seq', args.seq)
const seq = SessionSeq(args.seq)
const caller = this.callerOf(exec)
const sessionId = workspaceAccess.targetId(args, caller)
await workspaceAccess.authorizeTarget(ctx, caller, sessionId, exec.signal)
const trace = await serviceBoundary.call(ctx, exec.signal, 'event trace', () =>
ctx.sessionQuery.traceEvent({ sessionId, seq }, exec.signal))
workspaceAccess.assertObservedTargetAuthorized(caller, sessionId, trace.session)
const title = await workspaceAccess.readTitle(ctx, caller, sessionId, exec.signal)
return presentation.formatEventTrace(sessionId, title, trace)
}
/**
* Read an event window after authorizing its owning Session.
* @param args - Target event and surrounding event counts.
* @param exec - Agent-bound caller and cancellation signal.
* @returns Formatted event window.
*/
async executeEventRead(
args: EventReadArgs,
exec: ToolRunContext,
): Promise<string> {
const ctx = this.ctx
toolInput.assertNonNegativeSafeInteger('seq', args.seq)
const seq = SessionSeq(args.seq)
if (args.before !== undefined) toolInput.assertNonNegativeSafeInteger('before', args.before)
if (args.after !== undefined) toolInput.assertNonNegativeSafeInteger('after', args.after)
const caller = this.callerOf(exec)
const sessionId = workspaceAccess.targetId(args, caller)
await workspaceAccess.authorizeTarget(ctx, caller, sessionId, exec.signal)
const window = await serviceBoundary.call(ctx, exec.signal, 'event read', () =>
ctx.sessionQuery.readEvent({
sessionId,
seq,
...args.before === undefined ? {} : { before: args.before },
...args.after === undefined ? {} : { after: args.after },
}, exec.signal))
workspaceAccess.assertObservedTargetAuthorized(caller, sessionId, window.session)
const title = await workspaceAccess.readTitle(ctx, caller, sessionId, exec.signal)
return presentation.formatEventRead(sessionId, title, window)
}
async function executeEventRead(
ctx: Context,
turnBoundary: ProjectionHandle<TurnBoundaryProjection>,
args: EventReadArgs,
exec: ToolRunContext,
): Promise<string> {
toolInput.assertNonNegativeSafeInteger('seq', args.seq)
const seq = SessionSeq(args.seq)
if (args.before !== undefined) toolInput.assertNonNegativeSafeInteger('before', args.before)
if (args.after !== undefined) toolInput.assertNonNegativeSafeInteger('after', args.after)
const caller = workspaceAccess.callerOf(exec, turnBoundary)
const sessionId = workspaceAccess.targetId(args, caller)
await workspaceAccess.authorizeTarget(ctx, caller, sessionId, exec.signal)
const window = await serviceBoundary.call(ctx, exec.signal, 'event read', () =>
ctx.sessionQuery.readEvent({
sessionId,
seq,
...args.before === undefined ? {} : { before: args.before },
...args.after === undefined ? {} : { after: args.after },
}, exec.signal))
workspaceAccess.assertObservedTargetAuthorized(caller, sessionId, window.session)
const title = await workspaceAccess.readTitle(ctx, caller, sessionId, exec.signal)
return presentation.formatEventRead(sessionId, title, window)
}
async function collectPages<T>(
@@ -333,3 +276,12 @@ async function collectPages<T>(
cursor = page.nextCursor
}
}
/** Five model-facing session-query operation implementations. */
export const operations = {
executeSessionSearch,
executeEventSearch,
executeSessionTrace,
executeEventTrace,
executeEventRead,
}
@@ -6,6 +6,7 @@
import type { Context } from '@deepseek-ai/cordis'
import { brandString } from '@deepseek-ai/dsh-brand'
import { HarnessError } from '@deepseek-ai/dsh-llm'
import {
type SessionHeader,
type SessionId as SessionIdValue,
@@ -15,6 +16,8 @@ import type {
SessionLineageNode,
SessionRecord,
} from '@deepseek-ai/dsh-session-query'
import type { ToolRunContext } from '@deepseek-ai/dsh-tools'
import type { ProjectionHandle } from '@deepseek-ai/dsh-session-projection'
import { serviceBoundary } from './service-boundary.ts'
interface Caller {
@@ -50,6 +53,21 @@ interface DescendantVisit {
readonly next: DescendantVisit | undefined
}
function callerOf(exec: ToolRunContext, turnBoundary: ProjectionHandle<TurnBoundaryProjection>): Caller {
const agent = exec.agent
if (agent === undefined) {
throw new HarnessError(
'session query tools require an agent-bound caller',
'SESSION_QUERY_TOOL_MISSING_AGENT',
)
}
return {
id: agent.session.id,
header: agent.session.header,
boundary: turnBoundary.stateOf(agent.session),
}
}
function targetId(args: { readonly session_id?: string }, caller: Caller): SessionIdValue {
return args.session_id === undefined ? caller.id : brandString<SessionIdValue>(args.session_id)
}
@@ -224,6 +242,7 @@ function titleText(view: TitleView): string {
/** Workspace-scoped caller authorization, title access, and lineage projection. */
export const workspaceAccess = {
callerOf,
targetId,
authorizeTarget,
recordAuthorized,