diff --git a/.changeset/20261019143019-regenerate-sdk-from-openapi.md b/.changeset/20261019143019-regenerate-sdk-from-openapi.md new file mode 100644 index 000000000..efd8ff00f --- /dev/null +++ b/.changeset/20261019143019-regenerate-sdk-from-openapi.md @@ -0,0 +1,5 @@ +--- +"@truefoundry/trueforge-sdk": patch +--- + +Regenerate SDK from updated OpenAPI spec. diff --git a/.changeset/abandoned-turn-closer.md b/.changeset/abandoned-turn-closer.md new file mode 100644 index 000000000..812bdf6e5 --- /dev/null +++ b/.changeset/abandoned-turn-closer.md @@ -0,0 +1,6 @@ +--- +'@truefoundry/trueforge-core': patch +'@truefoundry/trueforge': patch +--- + +Close stale running turns via a controller loop and internal list/close endpoints (abandoned reason). diff --git a/.github/fern/openapi/openapi.json b/.github/fern/openapi/openapi.json index e99bf663c..6f9427a69 100644 --- a/.github/fern/openapi/openapi.json +++ b/.github/fern/openapi/openapi.json @@ -2,6 +2,32 @@ "components": { "parameters": {}, "schemas": { + "AbandonedTurn": { + "additionalProperties": false, + "properties": { + "session_id": { + "description": "Session that owns the turn.", + "minLength": 1, + "type": "string" + }, + "tenant_id": { + "description": "Tenant that owns the session.", + "minLength": 1, + "type": "string" + }, + "turn_id": { + "description": "Running turn with stale last activity.", + "minLength": 1, + "type": "string" + } + }, + "required": [ + "tenant_id", + "session_id", + "turn_id" + ], + "type": "object" + }, "ActionRequired": { "properties": { "id": { @@ -861,6 +887,32 @@ ], "type": "object" }, + "CloseAbandonedTurnRequest": { + "additionalProperties": false, + "properties": { + "session_id": { + "description": "Session that owns the turn.", + "minLength": 1, + "type": "string" + }, + "tenant_id": { + "description": "Tenant that owns the session.", + "minLength": 1, + "type": "string" + }, + "turn_id": { + "description": "Turn to close.", + "minLength": 1, + "type": "string" + } + }, + "required": [ + "tenant_id", + "session_id", + "turn_id" + ], + "type": "object" + }, "CompactionConfig": { "additionalProperties": false, "default": { @@ -1890,6 +1942,21 @@ }, "type": "object" }, + "ListAbandonedTurnsResponse": { + "additionalProperties": false, + "properties": { + "data": { + "items": { + "$ref": "#/components/schemas/AbandonedTurn" + }, + "type": "array" + } + }, + "required": [ + "data" + ], + "type": "object" + }, "ListAgentsResponse": { "properties": { "data": { @@ -6700,6 +6767,93 @@ "x-fern-sdk-method-name": "get_or_create_by_external_id" } }, + "/api/internal/turns/abandoned": { + "get": { + "description": "Returns all running turns whose last activity is past the stale threshold.", + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ListAbandonedTurnsResponse" + } + } + }, + "description": "Stale running turns oldest-first." + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RequestErrorResponse" + } + } + }, + "description": "Invalid service credential." + } + }, + "summary": "List abandoned running turns", + "tags": [ + "Internal" + ], + "x-excluded": true, + "x-fern-sdk-group-name": [ + "internal", + "turns" + ], + "x-fern-sdk-method-name": "list_abandoned" + } + }, + "/api/internal/turns/close-abandoned": { + "post": { + "description": "Cancels a stale running turn (turn.done with abandoned reason).", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CloseAbandonedTurnRequest" + } + } + }, + "required": true + }, + "responses": { + "204": { + "description": "Closed (or already terminal / missing turn)." + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RequestErrorResponse" + } + } + }, + "description": "Invalid service credential." + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RequestErrorResponse" + } + } + }, + "description": "Session not found." + } + }, + "summary": "Close an abandoned running turn", + "tags": [ + "Internal" + ], + "x-excluded": true, + "x-fern-sdk-group-name": [ + "internal", + "turns" + ], + "x-fern-sdk-method-name": "close_abandoned" + } + }, "/api/v1/agents": { "get": { "description": "List configured agents for the tenant, ordered by name. Optional `agent_name` filters by substring.", diff --git a/charts/trueforge/templates/controller-deployment.yaml b/charts/trueforge/templates/controller-deployment.yaml index 50238c894..56c40bee6 100644 --- a/charts/trueforge/templates/controller-deployment.yaml +++ b/charts/trueforge/templates/controller-deployment.yaml @@ -44,7 +44,7 @@ spec: priorityClassName: {{ . | quote }} {{- end }} # Above the controller's GRACEFUL_TIMEOUT_SECONDS so SIGKILL does not cut off loop drain. - terminationGracePeriodSeconds: {{ add .Values.server.gracefulTimeoutSeconds 5 }} + terminationGracePeriodSeconds: {{ .Values.server.terminationGracePeriodSeconds }} {{- with (include "trueforge.customCA.initContainer" .) }} initContainers: {{- . | nindent 8 }} diff --git a/charts/trueforge/templates/deployment.yaml b/charts/trueforge/templates/deployment.yaml index fa3fbfcf8..052bd2b6b 100644 --- a/charts/trueforge/templates/deployment.yaml +++ b/charts/trueforge/templates/deployment.yaml @@ -45,7 +45,7 @@ spec: priorityClassName: {{ . | quote }} {{- end }} # Above the server's GRACEFUL_TIMEOUT_SECONDS so SIGKILL does not cut off turn drain. - terminationGracePeriodSeconds: {{ add .Values.server.gracefulTimeoutSeconds 5 }} + terminationGracePeriodSeconds: {{ .Values.server.terminationGracePeriodSeconds }} {{- $caInit := include "trueforge.customCA.initContainer" . }} {{- if or $caInit .Values.initContainers }} initContainers: diff --git a/charts/trueforge/values.yaml b/charts/trueforge/values.yaml index 386e2cd7c..f16609032 100644 --- a/charts/trueforge/values.yaml +++ b/charts/trueforge/values.yaml @@ -120,6 +120,9 @@ server: publicBaseUrl: "" # Seconds the server is given to drain turns on SIGTERM (GRACEFUL_TIMEOUT_SECONDS). gracefulTimeoutSeconds: 30 + # Seconds Kubernetes waits after SIGTERM before SIGKILL. Must exceed + # gracefulTimeoutSeconds so the process can finish draining. + terminationGracePeriodSeconds: 300 # The server always runs peered (STANDALONE=false) against Postgres + Redis. # Escape hatch for knobs not covered by configs.*. Each entry is a full env # item: {name, value} or {name, valueFrom: {secretKeyRef|configMapKeyRef: ...}}. diff --git a/docs/openapi.json b/docs/openapi.json index e99bf663c..6f9427a69 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -2,6 +2,32 @@ "components": { "parameters": {}, "schemas": { + "AbandonedTurn": { + "additionalProperties": false, + "properties": { + "session_id": { + "description": "Session that owns the turn.", + "minLength": 1, + "type": "string" + }, + "tenant_id": { + "description": "Tenant that owns the session.", + "minLength": 1, + "type": "string" + }, + "turn_id": { + "description": "Running turn with stale last activity.", + "minLength": 1, + "type": "string" + } + }, + "required": [ + "tenant_id", + "session_id", + "turn_id" + ], + "type": "object" + }, "ActionRequired": { "properties": { "id": { @@ -861,6 +887,32 @@ ], "type": "object" }, + "CloseAbandonedTurnRequest": { + "additionalProperties": false, + "properties": { + "session_id": { + "description": "Session that owns the turn.", + "minLength": 1, + "type": "string" + }, + "tenant_id": { + "description": "Tenant that owns the session.", + "minLength": 1, + "type": "string" + }, + "turn_id": { + "description": "Turn to close.", + "minLength": 1, + "type": "string" + } + }, + "required": [ + "tenant_id", + "session_id", + "turn_id" + ], + "type": "object" + }, "CompactionConfig": { "additionalProperties": false, "default": { @@ -1890,6 +1942,21 @@ }, "type": "object" }, + "ListAbandonedTurnsResponse": { + "additionalProperties": false, + "properties": { + "data": { + "items": { + "$ref": "#/components/schemas/AbandonedTurn" + }, + "type": "array" + } + }, + "required": [ + "data" + ], + "type": "object" + }, "ListAgentsResponse": { "properties": { "data": { @@ -6700,6 +6767,93 @@ "x-fern-sdk-method-name": "get_or_create_by_external_id" } }, + "/api/internal/turns/abandoned": { + "get": { + "description": "Returns all running turns whose last activity is past the stale threshold.", + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ListAbandonedTurnsResponse" + } + } + }, + "description": "Stale running turns oldest-first." + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RequestErrorResponse" + } + } + }, + "description": "Invalid service credential." + } + }, + "summary": "List abandoned running turns", + "tags": [ + "Internal" + ], + "x-excluded": true, + "x-fern-sdk-group-name": [ + "internal", + "turns" + ], + "x-fern-sdk-method-name": "list_abandoned" + } + }, + "/api/internal/turns/close-abandoned": { + "post": { + "description": "Cancels a stale running turn (turn.done with abandoned reason).", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CloseAbandonedTurnRequest" + } + } + }, + "required": true + }, + "responses": { + "204": { + "description": "Closed (or already terminal / missing turn)." + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RequestErrorResponse" + } + } + }, + "description": "Invalid service credential." + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RequestErrorResponse" + } + } + }, + "description": "Session not found." + } + }, + "summary": "Close an abandoned running turn", + "tags": [ + "Internal" + ], + "x-excluded": true, + "x-fern-sdk-group-name": [ + "internal", + "turns" + ], + "x-fern-sdk-method-name": "close_abandoned" + } + }, "/api/v1/agents": { "get": { "description": "List configured agents for the tenant, ordered by name. Optional `agent_name` filters by substring.", diff --git a/packages/trueforge-core/src/agent-session/index.ts b/packages/trueforge-core/src/agent-session/index.ts index 1d05a2160..f373097d7 100644 --- a/packages/trueforge-core/src/agent-session/index.ts +++ b/packages/trueforge-core/src/agent-session/index.ts @@ -95,6 +95,7 @@ export type { InsertTurnInboundEventsInput, ListSessionEventsInput, ListSessionsInput, + ListStaleRunningTurnsInput, ListTurnEventsInput, ListTurnsInput, NewThreadInit, @@ -103,6 +104,7 @@ export type { PatchSandboxInfoInput, PatchThreadCapabilityStateInput, RemoveThreadsInput, + StaleRunningTurnRef, TurnContextAppend, TurnRecordWithoutSnapshot, UpdateSessionInput, diff --git a/packages/trueforge-core/src/agent-session/schemas/turn.ts b/packages/trueforge-core/src/agent-session/schemas/turn.ts index 700e44787..6f2005f37 100644 --- a/packages/trueforge-core/src/agent-session/schemas/turn.ts +++ b/packages/trueforge-core/src/agent-session/schemas/turn.ts @@ -18,7 +18,7 @@ export enum CancellationReason { ClientCancelled = 'client-cancelled', // Prior turn aborted because the client started a new turn. CancelledForNextTurn = 'cancelled-for-next-turn', - // Process shutting down (SIGTERM/SIGINT). + // Process shutting down (SIGTERM/SIGINT), or controller closed a stale running turn. Abandoned = 'abandoned', } diff --git a/packages/trueforge-core/src/agent-session/store/ISessionStore.ts b/packages/trueforge-core/src/agent-session/store/ISessionStore.ts index bdf8bed50..c0b289001 100644 --- a/packages/trueforge-core/src/agent-session/store/ISessionStore.ts +++ b/packages/trueforge-core/src/agent-session/store/ISessionStore.ts @@ -160,6 +160,18 @@ export interface ListTurnsInput { page_token: string | undefined; } +/** Input for listing running turns whose last activity is older than a cutoff. */ +export interface ListStaleRunningTurnsInput { + older_than: Date; +} + +/** Session/turn id pair for a stale running turn (tenant needed to load the session). */ +export interface StaleRunningTurnRef { + tenant_id: string; + session_id: string; + turn_id: string; +} + interface TurnStateUpdateKeys { session_id: string; turn_id: string; @@ -371,6 +383,13 @@ export interface ISessionStore< input: ListTurnsInput, ): Promise<{ data: TurnRecordWithoutSnapshot[]; pagination: TokenPagination }>; + /** + * All `running` turns whose last activity is strictly before `older_than`. + * Last activity is `MAX(session_event.created_at)`, or `turn.created_at` when + * the turn has no events. Oldest-first; no limit (caller batches). + */ + listStaleRunningTurns(input: ListStaleRunningTurnsInput): Promise; + /** Atomically performs `running ↔ paused` and appends turn.update. */ updateTurnNonTerminalState(input: UpdateTurnNonTerminalStateInput): Promise; diff --git a/packages/trueforge-core/src/agent-session/store/InMemorySessionStore.ts b/packages/trueforge-core/src/agent-session/store/InMemorySessionStore.ts index 4f11e3e8f..a7fa3da83 100644 --- a/packages/trueforge-core/src/agent-session/store/InMemorySessionStore.ts +++ b/packages/trueforge-core/src/agent-session/store/InMemorySessionStore.ts @@ -22,6 +22,7 @@ import type { ISessionStore, ListSessionEventsInput, ListSessionsInput, + ListStaleRunningTurnsInput, ListTurnEventsInput, ListTurnsInput, NewThreadInit, @@ -30,6 +31,7 @@ import type { PatchSandboxInfoInput, PatchThreadCapabilityStateInput, RemoveThreadsInput, + StaleRunningTurnRef, TurnContextAppend, TurnRecordWithoutSnapshot, UpdateSessionInput, @@ -479,6 +481,47 @@ export class InMemorySessionStore< return paginate(records, input.limit, input.page_token); } + async listStaleRunningTurns(input: ListStaleRunningTurnsInput): Promise { + const cutoffMs = input.older_than.getTime(); + const candidates: { + tenant_id: string; + session_id: string; + turn_id: string; + last_activity_ms: number; + }[] = []; + for (const turn of this.turns.values()) { + if (turn.state.status !== 'running') { + continue; + } + const session = this.sessions.get(sessionKey(turn.session_id)); + if (!session) { + continue; + } + const list = this.events.get(turnKey({ session_id: turn.session_id, turn_id: turn.turn_id })) ?? []; + const lastActivityMs = + list.length === 0 ? turn.created_at.getTime() : Math.max(...list.map(event => Date.parse(event.created_at))); + if (lastActivityMs < cutoffMs) { + candidates.push({ + tenant_id: session.record.tenant_id, + session_id: turn.session_id, + turn_id: turn.turn_id, + last_activity_ms: lastActivityMs, + }); + } + } + candidates.sort((a, b) => { + if (a.last_activity_ms !== b.last_activity_ms) { + return a.last_activity_ms - b.last_activity_ms; + } + const bySession = compareSessionId(a.session_id, b.session_id); + if (bySession !== 0) { + return bySession; + } + return a.turn_id < b.turn_id ? -1 : a.turn_id > b.turn_id ? 1 : 0; + }); + return candidates.map(({ tenant_id, session_id, turn_id }) => ({ tenant_id, session_id, turn_id })); + } + async updateTurnNonTerminalState(input: UpdateTurnNonTerminalStateInput): Promise { // Same as createTurn: synchronous body ⇒ atomic under run-to-completion. const tKey = turnKey(input); diff --git a/packages/trueforge-core/tests/agent-session/store/storeContractSuite.ts b/packages/trueforge-core/tests/agent-session/store/storeContractSuite.ts index 48532d783..ac1e555d5 100644 --- a/packages/trueforge-core/tests/agent-session/store/storeContractSuite.ts +++ b/packages/trueforge-core/tests/agent-session/store/storeContractSuite.ts @@ -3281,6 +3281,74 @@ export function runStoreContractSuite(createStore: () => ISessionStore) { expect(page2.data.map(t => t.turn_id)).toEqual(['t3']); }); + it('listStaleRunningTurns returns only running turns with stale last event', async () => { + const store = createStore(); + await seedSession(store); + const olderThan = new Date('2020-01-02T00:00:00.000Z'); + const staleEventAt = '2020-01-01T00:00:00.000Z'; + const freshEventAt = '2020-01-03T00:00:00.000Z'; + + await store.createTurn(makeCreateTurnInput({ sessionId, turnId: 'stale-running' })); + await store.appendToEvents({ + session_id: sessionId, + turn_id: 'stale-running', + events: [{ ...makeTurnCreatedEvent('stale-running'), created_at: staleEventAt }], + }); + + await store.createTurn( + makeCreateTurnInput({ + sessionId, + turnId: 'fresh-running', + previousTurnId: null, + firstTurnId: 'fresh-running', + }), + ); + await store.appendToEvents({ + session_id: sessionId, + turn_id: 'fresh-running', + events: [{ ...makeTurnCreatedEvent('fresh-running'), created_at: freshEventAt }], + }); + + await store.createTurn( + makeCreateTurnInput({ + sessionId, + turnId: 'stale-paused', + previousTurnId: null, + firstTurnId: 'stale-paused', + }), + ); + await store.appendToEvents({ + session_id: sessionId, + turn_id: 'stale-paused', + events: [{ ...makeTurnCreatedEvent('stale-paused'), created_at: staleEventAt }], + }); + const pausedState = makePausedTurnState(); + await store.updateTurnNonTerminalState({ + session_id: sessionId, + turn_id: 'stale-paused', + state: pausedState, + turn_update_event: makeTurnUpdateEvent(pausedState), + }); + + await store.createTurn( + makeCreateTurnInput({ + sessionId, + turnId: 'stale-done', + previousTurnId: null, + firstTurnId: 'stale-done', + }), + ); + await store.appendToEvents({ + session_id: sessionId, + turn_id: 'stale-done', + events: [{ ...makeTurnCreatedEvent('stale-done'), created_at: staleEventAt }], + }); + await finishTurn(store, 'stale-done'); + + const stale = await store.listStaleRunningTurns({ older_than: olderThan }); + expect(stale).toEqual([{ tenant_id: tenant, session_id: sessionId, turn_id: 'stale-running' }]); + }); + it('listTurnEvents supports desc order', async () => { const store = createStore(); await seedSession(store); diff --git a/packages/trueforge-sdk/package.json b/packages/trueforge-sdk/package.json index 5b75d6f84..9cf5e4d93 100644 --- a/packages/trueforge-sdk/package.json +++ b/packages/trueforge-sdk/package.json @@ -253,6 +253,17 @@ }, "default": "./dist/cjs/api/resources/internal/resources/sessions/exports.js" }, + "./internal/turns": { + "import": { + "types": "./dist/esm/api/resources/internal/resources/turns/exports.d.mts", + "default": "./dist/esm/api/resources/internal/resources/turns/exports.mjs" + }, + "require": { + "types": "./dist/cjs/api/resources/internal/resources/turns/exports.d.ts", + "default": "./dist/cjs/api/resources/internal/resources/turns/exports.js" + }, + "default": "./dist/cjs/api/resources/internal/resources/turns/exports.js" + }, "./internal/agents": { "import": { "types": "./dist/esm/api/resources/internal/resources/agents/exports.d.mts", diff --git a/packages/trueforge-sdk/reference.md b/packages/trueforge-sdk/reference.md index 185a73c8b..30944b8e4 100644 --- a/packages/trueforge-sdk/reference.md +++ b/packages/trueforge-sdk/reference.md @@ -3701,6 +3701,129 @@ await client.internal.sessions.getOrCreateByExternalId({ + + + + +## Internal Turns +
client.internal.turns.listAbandoned() -> TrueForge.ListAbandonedTurnsResponse +
+
+ +#### 📝 Description + +
+
+ +
+
+ +Returns all running turns whose last activity is past the stale threshold. +
+
+
+
+ +#### 🔌 Usage + +
+
+ +
+
+ +```typescript +await client.internal.turns.listAbandoned(); + +``` +
+
+
+
+ +#### ⚙️ Parameters + +
+
+ +
+
+ +**requestOptions:** `TurnsClient.RequestOptions` + +
+
+
+
+ + +
+
+
+ +
client.internal.turns.closeAbandoned({ ...params }) -> void +
+
+ +#### 📝 Description + +
+
+ +
+
+ +Cancels a stale running turn (turn.done with abandoned reason). +
+
+
+
+ +#### 🔌 Usage + +
+
+ +
+
+ +```typescript +await client.internal.turns.closeAbandoned({ + sessionId: "session_id", + tenantId: "tenant_id", + turnId: "turn_id" +}); + +``` +
+
+
+
+ +#### ⚙️ Parameters + +
+
+ +
+
+ +**request:** `TrueForge.internal.CloseAbandonedTurnRequest` + +
+
+ +
+
+ +**requestOptions:** `TurnsClient.RequestOptions` + +
+
+
+
+ +
diff --git a/packages/trueforge-sdk/src/api/resources/internal/client/Client.ts b/packages/trueforge-sdk/src/api/resources/internal/client/Client.ts index a80d41722..3eac70ac4 100644 --- a/packages/trueforge-sdk/src/api/resources/internal/client/Client.ts +++ b/packages/trueforge-sdk/src/api/resources/internal/client/Client.ts @@ -14,6 +14,7 @@ import { MetricsClient } from "../resources/metrics/client/Client.js"; import { SandboxEnvironmentsClient } from "../resources/sandboxEnvironments/client/Client.js"; import { SchedulesClient } from "../resources/schedules/client/Client.js"; import { SessionsClient } from "../resources/sessions/client/Client.js"; +import { TurnsClient } from "../resources/turns/client/Client.js"; export declare namespace InternalClient { export type Options = BaseClientOptions; @@ -27,6 +28,7 @@ export class InternalClient { protected _sandboxEnvironments: SandboxEnvironmentsClient | undefined; protected _schedules: SchedulesClient | undefined; protected _sessions: SessionsClient | undefined; + protected _turns: TurnsClient | undefined; protected _agents: AgentsClient | undefined; constructor(options: InternalClient.Options) { @@ -49,6 +51,10 @@ export class InternalClient { return (this._sessions ??= new SessionsClient(this._options)); } + public get turns(): TurnsClient { + return (this._turns ??= new TurnsClient(this._options)); + } + public get agents(): AgentsClient { return (this._agents ??= new AgentsClient(this._options)); } diff --git a/packages/trueforge-sdk/src/api/resources/internal/resources/index.ts b/packages/trueforge-sdk/src/api/resources/internal/resources/index.ts index 3c1662cf9..4f7cffd1e 100644 --- a/packages/trueforge-sdk/src/api/resources/internal/resources/index.ts +++ b/packages/trueforge-sdk/src/api/resources/internal/resources/index.ts @@ -8,3 +8,5 @@ export * from "./schedules/client/requests/index.js"; export * as schedules from "./schedules/index.js"; export * from "./sessions/client/requests/index.js"; export * as sessions from "./sessions/index.js"; +export * from "./turns/client/requests/index.js"; +export * as turns from "./turns/index.js"; diff --git a/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/Client.ts b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/Client.ts new file mode 100644 index 000000000..c51f752ef --- /dev/null +++ b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/Client.ts @@ -0,0 +1,211 @@ +// This file was auto-generated by Fern from our API Definition. + +import type { BaseClientOptions, BaseRequestOptions } from "../../../../../../BaseClient.js"; +import { type NormalizedClientOptionsWithAuth, normalizeClientOptionsWithAuth } from "../../../../../../BaseClient.js"; +import { mergeHeaders } from "../../../../../../core/headers.js"; +import * as core from "../../../../../../core/index.js"; +import { mergeAdditionalBodyParameters } from "../../../../../../core/requestBody.js"; +import { handleNonStatusCodeError } from "../../../../../../errors/handleNonStatusCodeError.js"; +import * as errors from "../../../../../../errors/index.js"; +import * as serializers from "../../../../../../serialization/index.js"; +import * as TrueForge from "../../../../../index.js"; + +export declare namespace TurnsClient { + export type Options = BaseClientOptions; + + export interface RequestOptions extends BaseRequestOptions {} +} + +export class TurnsClient { + protected readonly _options: NormalizedClientOptionsWithAuth; + + constructor(options: TurnsClient.Options) { + this._options = normalizeClientOptionsWithAuth(options); + } + + /** + * Returns all running turns whose last activity is past the stale threshold. + * + * @param {TurnsClient.RequestOptions} requestOptions - Request-specific configuration. + * + * @throws {@link TrueForge.UnauthorizedError} + * @throws {@link errors.TrueForgeError} + * @throws {@link errors.TrueForgeTimeoutError} + * + * @example + * await client.internal.turns.listAbandoned() + */ + public listAbandoned( + requestOptions?: TurnsClient.RequestOptions, + ): core.HttpResponsePromise { + return core.HttpResponsePromise.fromPromise(this.__listAbandoned(requestOptions)); + } + + private async __listAbandoned( + requestOptions?: TurnsClient.RequestOptions, + ): Promise> { + const _authRequest: core.AuthRequest = await this._options.authProvider.getAuthRequest(); + const _headers: core.Fetcher.Args["headers"] = mergeHeaders( + _authRequest.headers, + this._options?.headers, + requestOptions?.headers, + ); + const _response = await (this._options.fetcher ?? core.fetcher)({ + url: core.url.join( + (await core.Supplier.get(this._options.baseUrl)) ?? + (await core.Supplier.get(this._options.environment)), + "api/internal/turns/abandoned", + ), + method: "GET", + headers: _headers, + queryString: core.url.queryBuilder().mergeAdditional(requestOptions?.queryParams).build(), + timeoutMs: (requestOptions?.timeoutInSeconds ?? this._options?.timeoutInSeconds ?? 60) * 1000, + maxRetries: requestOptions?.maxRetries ?? this._options?.maxRetries, + abortSignal: requestOptions?.abortSignal, + fetchFn: this._options?.fetch, + logging: this._options.logging, + }); + if (_response.ok) { + return { + data: serializers.ListAbandonedTurnsResponse.parseOrThrow(_response.body, { + unrecognizedObjectKeys: "passthrough", + allowUnrecognizedUnionMembers: true, + allowUnrecognizedEnumValues: true, + skipValidation: true, + breadcrumbsPrefix: ["response"], + }), + rawResponse: _response.rawResponse, + }; + } + + if (_response.error.reason === "status-code") { + switch (_response.error.statusCode) { + case 401: + throw new TrueForge.UnauthorizedError( + serializers.RequestErrorResponse.parseOrThrow(_response.error.body, { + unrecognizedObjectKeys: "passthrough", + allowUnrecognizedUnionMembers: true, + allowUnrecognizedEnumValues: true, + skipValidation: true, + breadcrumbsPrefix: ["response"], + }), + _response.rawResponse, + ); + default: + throw new errors.TrueForgeError({ + statusCode: _response.error.statusCode, + body: _response.error.body, + rawResponse: _response.rawResponse, + }); + } + } + + return handleNonStatusCodeError(_response.error, _response.rawResponse, "GET", "/api/internal/turns/abandoned"); + } + + /** + * Cancels a stale running turn (turn.done with abandoned reason). + * + * @param {TrueForge.internal.CloseAbandonedTurnRequest} request + * @param {TurnsClient.RequestOptions} requestOptions - Request-specific configuration. + * + * @throws {@link TrueForge.UnauthorizedError} + * @throws {@link TrueForge.NotFoundError} + * @throws {@link errors.TrueForgeError} + * @throws {@link errors.TrueForgeTimeoutError} + * + * @example + * await client.internal.turns.closeAbandoned({ + * sessionId: "session_id", + * tenantId: "tenant_id", + * turnId: "turn_id" + * }) + */ + public closeAbandoned( + request: TrueForge.internal.CloseAbandonedTurnRequest, + requestOptions?: TurnsClient.RequestOptions, + ): core.HttpResponsePromise { + return core.HttpResponsePromise.fromPromise(this.__closeAbandoned(request, requestOptions)); + } + + private async __closeAbandoned( + request: TrueForge.internal.CloseAbandonedTurnRequest, + requestOptions?: TurnsClient.RequestOptions, + ): Promise> { + const _authRequest: core.AuthRequest = await this._options.authProvider.getAuthRequest(); + const _headers: core.Fetcher.Args["headers"] = mergeHeaders( + _authRequest.headers, + this._options?.headers, + requestOptions?.headers, + ); + const _response = await (this._options.fetcher ?? core.fetcher)({ + url: core.url.join( + (await core.Supplier.get(this._options.baseUrl)) ?? + (await core.Supplier.get(this._options.environment)), + "api/internal/turns/close-abandoned", + ), + method: "POST", + headers: _headers, + contentType: "application/json", + queryString: core.url.queryBuilder().mergeAdditional(requestOptions?.queryParams).build(), + requestType: "json", + body: mergeAdditionalBodyParameters( + serializers.internal.CloseAbandonedTurnRequest.jsonOrThrow(request, { + unrecognizedObjectKeys: "passthrough", + allowUnrecognizedUnionMembers: true, + allowUnrecognizedEnumValues: true, + omitUndefined: true, + }), + requestOptions?.additionalBodyParameters, + ), + timeoutMs: (requestOptions?.timeoutInSeconds ?? this._options?.timeoutInSeconds ?? 60) * 1000, + maxRetries: requestOptions?.maxRetries ?? this._options?.maxRetries, + abortSignal: requestOptions?.abortSignal, + fetchFn: this._options?.fetch, + logging: this._options.logging, + }); + if (_response.ok) { + return { data: undefined, rawResponse: _response.rawResponse }; + } + + if (_response.error.reason === "status-code") { + switch (_response.error.statusCode) { + case 401: + throw new TrueForge.UnauthorizedError( + serializers.RequestErrorResponse.parseOrThrow(_response.error.body, { + unrecognizedObjectKeys: "passthrough", + allowUnrecognizedUnionMembers: true, + allowUnrecognizedEnumValues: true, + skipValidation: true, + breadcrumbsPrefix: ["response"], + }), + _response.rawResponse, + ); + case 404: + throw new TrueForge.NotFoundError( + serializers.RequestErrorResponse.parseOrThrow(_response.error.body, { + unrecognizedObjectKeys: "passthrough", + allowUnrecognizedUnionMembers: true, + allowUnrecognizedEnumValues: true, + skipValidation: true, + breadcrumbsPrefix: ["response"], + }), + _response.rawResponse, + ); + default: + throw new errors.TrueForgeError({ + statusCode: _response.error.statusCode, + body: _response.error.body, + rawResponse: _response.rawResponse, + }); + } + } + + return handleNonStatusCodeError( + _response.error, + _response.rawResponse, + "POST", + "/api/internal/turns/close-abandoned", + ); + } +} diff --git a/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/index.ts b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/index.ts new file mode 100644 index 000000000..195f9aa8a --- /dev/null +++ b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/index.ts @@ -0,0 +1 @@ +export * from "./requests/index.js"; diff --git a/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/requests/CloseAbandonedTurnRequest.ts b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/requests/CloseAbandonedTurnRequest.ts new file mode 100644 index 000000000..ccecf290b --- /dev/null +++ b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/requests/CloseAbandonedTurnRequest.ts @@ -0,0 +1,18 @@ +// This file was auto-generated by Fern from our API Definition. + +/** + * @example + * { + * sessionId: "session_id", + * tenantId: "tenant_id", + * turnId: "turn_id" + * } + */ +export interface CloseAbandonedTurnRequest { + /** Session that owns the turn. */ + sessionId: string; + /** Tenant that owns the session. */ + tenantId: string; + /** Turn to close. */ + turnId: string; +} diff --git a/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/requests/index.ts b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/requests/index.ts new file mode 100644 index 000000000..05220cf4d --- /dev/null +++ b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/client/requests/index.ts @@ -0,0 +1 @@ +export type { CloseAbandonedTurnRequest } from "./CloseAbandonedTurnRequest.js"; diff --git a/packages/trueforge-sdk/src/api/resources/internal/resources/turns/exports.ts b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/exports.ts new file mode 100644 index 000000000..530dd7d3d --- /dev/null +++ b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/exports.ts @@ -0,0 +1,4 @@ +// This file was auto-generated by Fern from our API Definition. + +export { TurnsClient } from "./client/Client.js"; +export * from "./client/index.js"; diff --git a/packages/trueforge-sdk/src/api/resources/internal/resources/turns/index.ts b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/index.ts new file mode 100644 index 000000000..914b8c3c7 --- /dev/null +++ b/packages/trueforge-sdk/src/api/resources/internal/resources/turns/index.ts @@ -0,0 +1 @@ +export * from "./client/index.js"; diff --git a/packages/trueforge-sdk/src/api/types/AbandonedTurn.ts b/packages/trueforge-sdk/src/api/types/AbandonedTurn.ts new file mode 100644 index 000000000..e85e8ee2c --- /dev/null +++ b/packages/trueforge-sdk/src/api/types/AbandonedTurn.ts @@ -0,0 +1,10 @@ +// This file was auto-generated by Fern from our API Definition. + +export interface AbandonedTurn { + /** Session that owns the turn. */ + sessionId: string; + /** Tenant that owns the session. */ + tenantId: string; + /** Running turn with stale last activity. */ + turnId: string; +} diff --git a/packages/trueforge-sdk/src/api/types/ListAbandonedTurnsResponse.ts b/packages/trueforge-sdk/src/api/types/ListAbandonedTurnsResponse.ts new file mode 100644 index 000000000..14b8f59db --- /dev/null +++ b/packages/trueforge-sdk/src/api/types/ListAbandonedTurnsResponse.ts @@ -0,0 +1,7 @@ +// This file was auto-generated by Fern from our API Definition. + +import type * as TrueForge from "../index.js"; + +export interface ListAbandonedTurnsResponse { + data: TrueForge.AbandonedTurn[]; +} diff --git a/packages/trueforge-sdk/src/api/types/index.ts b/packages/trueforge-sdk/src/api/types/index.ts index e05a49b08..fa4010c31 100644 --- a/packages/trueforge-sdk/src/api/types/index.ts +++ b/packages/trueforge-sdk/src/api/types/index.ts @@ -1,3 +1,4 @@ +export * from "./AbandonedTurn.js"; export * from "./ActionRequired.js"; export * from "./ActionRequiredEvent.js"; export * from "./Agent.js"; @@ -94,6 +95,7 @@ export * from "./GoogleGeminiModelProvider.js"; export * from "./InitialUserMessage.js"; export * from "./InputTokensCompactionTrigger.js"; export * from "./LargeToolResponseConfig.js"; +export * from "./ListAbandonedTurnsResponse.js"; export * from "./ListAgentsResponse.js"; export * from "./ListAvailableMcpServersResponse.js"; export * from "./ListAvailableModelsResponse.js"; diff --git a/packages/trueforge-sdk/src/serialization/resources/internal/resources/index.ts b/packages/trueforge-sdk/src/serialization/resources/internal/resources/index.ts index 0c87d3959..f5e1a7306 100644 --- a/packages/trueforge-sdk/src/serialization/resources/internal/resources/index.ts +++ b/packages/trueforge-sdk/src/serialization/resources/internal/resources/index.ts @@ -4,3 +4,5 @@ export * from "./schedules/client/requests/index.js"; export * as schedules from "./schedules/index.js"; export * from "./sessions/client/requests/index.js"; export * as sessions from "./sessions/index.js"; +export * from "./turns/client/requests/index.js"; +export * as turns from "./turns/index.js"; diff --git a/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/client/index.ts b/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/client/index.ts new file mode 100644 index 000000000..195f9aa8a --- /dev/null +++ b/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/client/index.ts @@ -0,0 +1 @@ +export * from "./requests/index.js"; diff --git a/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/client/requests/CloseAbandonedTurnRequest.ts b/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/client/requests/CloseAbandonedTurnRequest.ts new file mode 100644 index 000000000..aeb89f8cd --- /dev/null +++ b/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/client/requests/CloseAbandonedTurnRequest.ts @@ -0,0 +1,22 @@ +// This file was auto-generated by Fern from our API Definition. + +import type * as TrueForge from "../../../../../../../api/index.js"; +import * as core from "../../../../../../../core/index.js"; +import type * as serializers from "../../../../../../index.js"; + +export const CloseAbandonedTurnRequest: core.serialization.Schema< + serializers.internal.CloseAbandonedTurnRequest.Raw, + TrueForge.internal.CloseAbandonedTurnRequest +> = core.serialization.object({ + sessionId: core.serialization.property("session_id", core.serialization.string()), + tenantId: core.serialization.property("tenant_id", core.serialization.string()), + turnId: core.serialization.property("turn_id", core.serialization.string()), +}); + +export declare namespace CloseAbandonedTurnRequest { + export interface Raw { + session_id: string; + tenant_id: string; + turn_id: string; + } +} diff --git a/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/client/requests/index.ts b/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/client/requests/index.ts new file mode 100644 index 000000000..9c8f1e270 --- /dev/null +++ b/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/client/requests/index.ts @@ -0,0 +1 @@ +export { CloseAbandonedTurnRequest } from "./CloseAbandonedTurnRequest.js"; diff --git a/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/index.ts b/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/index.ts new file mode 100644 index 000000000..914b8c3c7 --- /dev/null +++ b/packages/trueforge-sdk/src/serialization/resources/internal/resources/turns/index.ts @@ -0,0 +1 @@ +export * from "./client/index.js"; diff --git a/packages/trueforge-sdk/src/serialization/types/AbandonedTurn.ts b/packages/trueforge-sdk/src/serialization/types/AbandonedTurn.ts new file mode 100644 index 000000000..d0fb79370 --- /dev/null +++ b/packages/trueforge-sdk/src/serialization/types/AbandonedTurn.ts @@ -0,0 +1,20 @@ +// This file was auto-generated by Fern from our API Definition. + +import type * as TrueForge from "../../api/index.js"; +import * as core from "../../core/index.js"; +import type * as serializers from "../index.js"; + +export const AbandonedTurn: core.serialization.ObjectSchema = + core.serialization.object({ + sessionId: core.serialization.property("session_id", core.serialization.string()), + tenantId: core.serialization.property("tenant_id", core.serialization.string()), + turnId: core.serialization.property("turn_id", core.serialization.string()), + }); + +export declare namespace AbandonedTurn { + export interface Raw { + session_id: string; + tenant_id: string; + turn_id: string; + } +} diff --git a/packages/trueforge-sdk/src/serialization/types/ListAbandonedTurnsResponse.ts b/packages/trueforge-sdk/src/serialization/types/ListAbandonedTurnsResponse.ts new file mode 100644 index 000000000..b906d8e1b --- /dev/null +++ b/packages/trueforge-sdk/src/serialization/types/ListAbandonedTurnsResponse.ts @@ -0,0 +1,19 @@ +// This file was auto-generated by Fern from our API Definition. + +import type * as TrueForge from "../../api/index.js"; +import * as core from "../../core/index.js"; +import type * as serializers from "../index.js"; +import { AbandonedTurn } from "./AbandonedTurn.js"; + +export const ListAbandonedTurnsResponse: core.serialization.ObjectSchema< + serializers.ListAbandonedTurnsResponse.Raw, + TrueForge.ListAbandonedTurnsResponse +> = core.serialization.object({ + data: core.serialization.list(AbandonedTurn), +}); + +export declare namespace ListAbandonedTurnsResponse { + export interface Raw { + data: AbandonedTurn.Raw[]; + } +} diff --git a/packages/trueforge-sdk/src/serialization/types/index.ts b/packages/trueforge-sdk/src/serialization/types/index.ts index e05a49b08..fa4010c31 100644 --- a/packages/trueforge-sdk/src/serialization/types/index.ts +++ b/packages/trueforge-sdk/src/serialization/types/index.ts @@ -1,3 +1,4 @@ +export * from "./AbandonedTurn.js"; export * from "./ActionRequired.js"; export * from "./ActionRequiredEvent.js"; export * from "./Agent.js"; @@ -94,6 +95,7 @@ export * from "./GoogleGeminiModelProvider.js"; export * from "./InitialUserMessage.js"; export * from "./InputTokensCompactionTrigger.js"; export * from "./LargeToolResponseConfig.js"; +export * from "./ListAbandonedTurnsResponse.js"; export * from "./ListAgentsResponse.js"; export * from "./ListAvailableMcpServersResponse.js"; export * from "./ListAvailableModelsResponse.js"; diff --git a/packages/trueforge-sdk/tests/wire/internal/turns.test.ts b/packages/trueforge-sdk/tests/wire/internal/turns.test.ts new file mode 100644 index 000000000..bb6086fb4 --- /dev/null +++ b/packages/trueforge-sdk/tests/wire/internal/turns.test.ts @@ -0,0 +1,121 @@ +// This file was auto-generated by Fern from our API Definition. + +import * as TrueForgeTypes from "../../../src/api/index"; +import { TrueForge } from "../../../src/Client"; +import { mockServerPool } from "../../mock-server/MockServerPool"; + +describe("TurnsClient", () => { + test("list_abandoned (1)", async () => { + const server = mockServerPool.createServer(); + const client = new TrueForge({ maxRetries: 0, token: "test", baseUrl: server.baseUrl }); + + const rawResponseBody = { data: [{ session_id: "session_id", tenant_id: "tenant_id", turn_id: "turn_id" }] }; + + server + .mockEndpoint() + .get("/api/internal/turns/abandoned") + .respondWith() + .statusCode(200) + .jsonBody(rawResponseBody) + .build(); + + const response = await client.internal.turns.listAbandoned(); + expect(response).toEqual({ + data: [ + { + sessionId: "session_id", + tenantId: "tenant_id", + turnId: "turn_id", + }, + ], + }); + }); + + test("list_abandoned (2)", async () => { + const server = mockServerPool.createServer(); + const client = new TrueForge({ maxRetries: 0, token: "test", baseUrl: server.baseUrl }); + + const rawResponseBody = { error: { message: "message" } }; + + server + .mockEndpoint() + .get("/api/internal/turns/abandoned") + .respondWith() + .statusCode(401) + .jsonBody(rawResponseBody) + .build(); + + await expect(async () => { + return await client.internal.turns.listAbandoned(); + }).rejects.toThrow(TrueForgeTypes.UnauthorizedError); + }); + + test("close_abandoned (1)", async () => { + const server = mockServerPool.createServer(); + const client = new TrueForge({ maxRetries: 0, token: "test", baseUrl: server.baseUrl }); + const rawRequestBody = { session_id: "session_id", tenant_id: "tenant_id", turn_id: "turn_id" }; + + server + .mockEndpoint() + .post("/api/internal/turns/close-abandoned") + .jsonBody(rawRequestBody) + .respondWith() + .statusCode(200) + .build(); + + const response = await client.internal.turns.closeAbandoned({ + sessionId: "session_id", + tenantId: "tenant_id", + turnId: "turn_id", + }); + expect(response).toEqual(undefined); + }); + + test("close_abandoned (2)", async () => { + const server = mockServerPool.createServer(); + const client = new TrueForge({ maxRetries: 0, token: "test", baseUrl: server.baseUrl }); + const rawRequestBody = { session_id: "x", tenant_id: "x", turn_id: "x" }; + const rawResponseBody = { error: { message: "message" } }; + + server + .mockEndpoint() + .post("/api/internal/turns/close-abandoned") + .jsonBody(rawRequestBody) + .respondWith() + .statusCode(401) + .jsonBody(rawResponseBody) + .build(); + + await expect(async () => { + return await client.internal.turns.closeAbandoned({ + sessionId: "x", + tenantId: "x", + turnId: "x", + }); + }).rejects.toThrow(TrueForgeTypes.UnauthorizedError); + }); + + test("close_abandoned (3)", async () => { + const server = mockServerPool.createServer(); + const client = new TrueForge({ maxRetries: 0, token: "test", baseUrl: server.baseUrl }); + const rawRequestBody = { session_id: "x", tenant_id: "x", turn_id: "x" }; + const rawResponseBody = { error: { message: "message" } }; + + server + .mockEndpoint() + .post("/api/internal/turns/close-abandoned") + .jsonBody(rawRequestBody) + .respondWith() + .statusCode(404) + .jsonBody(rawResponseBody) + .build(); + + await expect(async () => { + return await client.internal.turns.closeAbandoned({ + sessionId: "x", + tenantId: "x", + turnId: "x", + }); + }).rejects.toThrow(TrueForgeTypes.NotFoundError); + }); +}); diff --git a/packages/trueforge/.env.example b/packages/trueforge/.env.example index 684ed8de2..68bb9825f 100644 --- a/packages/trueforge/.env.example +++ b/packages/trueforge/.env.example @@ -129,6 +129,14 @@ REDIS_URL=redis://localhost:6379 ## Defaults to http://localhost:$PORT; point it at the server Service in a split deployment. # SERVER_URL=http://localhost:8790 +## Abandoned-turn closer (controller + internal /api/internal/turns). All optional. +## Running turns with last activity older than this are closed. Default 900000 (15 minutes). +# ABANDONED_TURN_STALE_MS=900000 +## Max closes per controller tick. Default 50. +# ABANDONED_TURN_CLOSE_BATCH=50 +## Gap between closer loop passes. Default 60000. +# ABANDONED_TURN_CLOSER_INTERVAL_MS=60000 + ## Service credential for controller → /api/internal/schedules. ## Required when STANDALONE=false. Standalone defaults to `trueforge-standalone` (local only). TRUEFORGE_API_KEY=placeholder-value-please-generate-your-own diff --git a/packages/trueforge/src/apis/abandonedTurns.ts b/packages/trueforge/src/apis/abandonedTurns.ts new file mode 100644 index 000000000..269e26d2b --- /dev/null +++ b/packages/trueforge/src/apis/abandonedTurns.ts @@ -0,0 +1,63 @@ +/** + * Internal API for the abandoned-turn closer controller. + */ +import { OpenAPIHono, type RouteHandler } from '@hono/zod-openapi'; +import { CancellationReason, type ISessionStore, type Sessions } from '@truefoundry/trueforge-core/agent-session'; +import type { RedisClient } from '@truefoundry/trueforge-core/request-reply'; +import type { Logger } from 'winston'; +import configuration from '../config'; +import { closeAbandonedTurnRoute, listAbandonedTurnsRoute } from '../routes/abandonedTurnRoutes'; +import type { ActiveTurnRegistry } from '../runtime/activeTurns'; +import { cancelSessionTurn } from './sessions'; + +export interface AbandonedTurnsRouterDeps { + sessionStore: ISessionStore; + sessions: Sessions; + activeTurns: ActiveTurnRegistry; + redis?: RedisClient | undefined; + logger: Logger; +} + +export function createAbandonedTurnsRouter(deps: AbandonedTurnsRouterDeps) { + const listHandler: RouteHandler = async c => { + const olderThan = new Date(Date.now() - configuration.ABANDONED_TURN_STALE_MS); + const stale = await deps.sessionStore.listStaleRunningTurns({ older_than: olderThan }); + return c.json( + { + data: stale.map(row => ({ + tenant_id: row.tenant_id, + session_id: row.session_id, + turn_id: row.turn_id, + })), + }, + 200, + ); + }; + + const closeHandler: RouteHandler = async c => { + const body = c.req.valid('json'); + const session = await deps.sessions.get({ + tenant_id: body.tenant_id, + session_id: body.session_id, + }); + if (!session) { + return c.json({ error: { message: `Session not found: ${body.session_id}` } }, 404); + } + await cancelSessionTurn( + { + activeTurns: deps.activeTurns, + session, + sessionStore: deps.sessionStore, + redis: deps.redis, + logger: deps.logger, + }, + { turnId: body.turn_id, reason: CancellationReason.Abandoned }, + ); + return c.body(null, 204); + }; + + const router = new OpenAPIHono(); + router.openapi(listAbandonedTurnsRoute, listHandler); + router.openapi(closeAbandonedTurnRoute, closeHandler); + return router; +} diff --git a/packages/trueforge/src/app.ts b/packages/trueforge/src/app.ts index b66bbab55..0b4234292 100644 --- a/packages/trueforge/src/app.ts +++ b/packages/trueforge/src/app.ts @@ -10,6 +10,7 @@ import { HTTPException } from 'hono/http-exception'; import { randomUUID } from 'node:crypto'; import type { Configuration } from 'openid-client'; import type { Logger } from 'winston'; +import { createAbandonedTurnsRouter } from './apis/abandonedTurns'; import { createAgentsRouter } from './apis/agents'; import { createAuthRouter } from './apis/auth'; import { createCapabilitiesRouter } from './apis/capabilities'; @@ -491,6 +492,19 @@ export function createServerApp(deps: ServerDeps) { scheduleExecutionAuthMiddleware, ), ); + app.route( + '/api/internal/turns', + withAuth( + createAbandonedTurnsRouter({ + sessionStore: deps.sessionStore, + sessions: deps.sessions, + activeTurns: deps.activeTurns, + redis: deps.redis, + logger: deps.logger, + }), + scheduleExecutionAuthMiddleware, + ), + ); app.route( '/api/v1/schedules', withAuth( diff --git a/packages/trueforge/src/config.ts b/packages/trueforge/src/config.ts index 2913f3551..4693ecff0 100644 --- a/packages/trueforge/src/config.ts +++ b/packages/trueforge/src/config.ts @@ -825,6 +825,21 @@ export interface SharedServerConfiguration { * Default: `http://localhost:$PORT`; in-cluster deployments MUST point this at the server Service. */ SERVER_URL: string; + /** + * Running turns with last activity older than this are closed as abandoned. + * Env: `ABANDONED_TURN_STALE_MS`. Default 900000 (15 minutes). + */ + ABANDONED_TURN_STALE_MS: number; + /** + * Max abandoned turns the controller closes per tick. + * Env: `ABANDONED_TURN_CLOSE_BATCH`. Default 50. + */ + ABANDONED_TURN_CLOSE_BATCH: number; + /** + * Gap between abandoned-turn closer loop passes. + * Env: `ABANDONED_TURN_CLOSER_INTERVAL_MS`. Default 60000. + */ + ABANDONED_TURN_CLOSER_INTERVAL_MS: number; /** * Service credential for controller calls. * Env: `TRUEFORGE_API_KEY`. Required and non-empty when `STANDALONE=false`. @@ -1199,6 +1214,21 @@ const shared: SharedServerConfiguration = { PUBLIC_BASE_URL: parsePublicBaseUrl(getEnv('PUBLIC_BASE_URL', { defaultValue: '' })), SERVER_URL: getEnv('SERVER_URL', { defaultValue: `http://localhost:${String(port)}` }) ?? `http://localhost:${String(port)}`, + ABANDONED_TURN_STALE_MS: parsePositiveInt({ + envKey: 'ABANDONED_TURN_STALE_MS', + raw: getEnv('ABANDONED_TURN_STALE_MS'), + defaultValue: 15 * 60_000, + }), + ABANDONED_TURN_CLOSE_BATCH: parsePositiveInt({ + envKey: 'ABANDONED_TURN_CLOSE_BATCH', + raw: getEnv('ABANDONED_TURN_CLOSE_BATCH'), + defaultValue: 50, + }), + ABANDONED_TURN_CLOSER_INTERVAL_MS: parsePositiveInt({ + envKey: 'ABANDONED_TURN_CLOSER_INTERVAL_MS', + raw: getEnv('ABANDONED_TURN_CLOSER_INTERVAL_MS'), + defaultValue: 60_000, + }), TRUEFORGE_API_KEY: standalone ? (getEnv('TRUEFORGE_API_KEY', { defaultValue: STANDALONE_TRUEFORGE_API_KEY }) ?? STANDALONE_TRUEFORGE_API_KEY) : (getEnv('TRUEFORGE_API_KEY', { required: true }) ?? ''), diff --git a/packages/trueforge/src/controller.ts b/packages/trueforge/src/controller.ts index 21e5958f6..b39784f05 100644 --- a/packages/trueforge/src/controller.ts +++ b/packages/trueforge/src/controller.ts @@ -1,4 +1,5 @@ import type { Logger } from 'winston'; +import { abandonedTurnCloserLoop } from './controller/abandonedTurnCloser'; import { Controller } from './controller/Controller'; import { sandboxEnvBuildLoop } from './controller/sandboxEnvBuild'; import { scheduleDispatchLoop } from './controller/scheduleDispatch'; @@ -9,7 +10,7 @@ import type { WithTransaction } from './db/transaction'; * Controller whose schedule loop hands runs to the server over HTTP * (`SERVER_URL` + `TRUEFORGE_API_KEY` from process config). Standalone uses * loopback; distributed uses the dedicated controller against the server Service. - * Sandbox-env build loop lists/progresses pending versions over the same HTTP path. + * Sandbox-env build and abandoned-turn closer loops use the same HTTP path. */ export function createController(params: { scheduleStore: IScheduleStore; @@ -24,6 +25,7 @@ export function createController(params: { logger: params.logger, }), sandboxEnvBuildLoop({ logger: params.logger }), + abandonedTurnCloserLoop({ logger: params.logger }), ], logger: params.logger, }); diff --git a/packages/trueforge/src/controller/abandonedTurnCloser.ts b/packages/trueforge/src/controller/abandonedTurnCloser.ts new file mode 100644 index 000000000..1d59b2b83 --- /dev/null +++ b/packages/trueforge/src/controller/abandonedTurnCloser.ts @@ -0,0 +1,79 @@ +/** + * Control loop: list stale running turns and close them over HTTP (batch per tick). + */ +import { TrueForgeApi } from '@truefoundry/trueforge-sdk'; +import type { Logger } from 'winston'; +import configuration from '../config'; +import { captureCriticalException } from '../sentry'; +import type { ControlLoop } from './Controller'; +import { internalTrueForgeClient } from './internalTrueForgeClient'; + +const ABANDONED_TURN_CLOSER_LOOP_NAME = 'abandoned-turn-closer'; + +export async function dispatchAbandonedTurns({ logger }: { logger: Logger }): Promise { + const abandoned = await internalTrueForgeClient.listAbandonedTurns(); + const batch = abandoned.slice(0, configuration.ABANDONED_TURN_CLOSE_BATCH); + logger.debug('Abandoned turn closer tick', { + abandoned_count: abandoned.length, + close_batch: batch.length, + }); + for (const turn of batch) { + logger.debug('Closing abandoned turn', { + tenant_id: turn.tenantId, + session_id: turn.sessionId, + turn_id: turn.turnId, + }); + try { + await internalTrueForgeClient.closeAbandonedTurn(turn); + logger.debug('Abandoned turn closed', { + tenant_id: turn.tenantId, + session_id: turn.sessionId, + turn_id: turn.turnId, + }); + } catch (error) { + if (error instanceof TrueForgeApi.NotFoundError) { + logger.warn('Abandoned turn session not found; skipping', { + tenant_id: turn.tenantId, + session_id: turn.sessionId, + turn_id: turn.turnId, + }); + captureCriticalException(error, { + tags: { module: 'abandonedTurnCloser', operation: 'closeNotFound' }, + extra: { + tenant_id: turn.tenantId, + session_id: turn.sessionId, + turn_id: turn.turnId, + }, + }); + continue; + } + logger.error('Abandoned turn close failed', { + tenant_id: turn.tenantId, + session_id: turn.sessionId, + turn_id: turn.turnId, + error: error instanceof Error ? error.message : String(error), + }); + captureCriticalException(error, { + tags: { module: 'abandonedTurnCloser', operation: 'close' }, + extra: { + tenant_id: turn.tenantId, + session_id: turn.sessionId, + turn_id: turn.turnId, + }, + }); + } + } +} + +export function abandonedTurnCloserLoop(params: { logger: Logger }): ControlLoop { + return { + name: ABANDONED_TURN_CLOSER_LOOP_NAME, + intervalMs: configuration.ABANDONED_TURN_CLOSER_INTERVAL_MS, + async tick(signal) { + if (signal.aborted) { + return; + } + await dispatchAbandonedTurns({ logger: params.logger }); + }, + }; +} diff --git a/packages/trueforge/src/controller/internalTrueForgeClient.ts b/packages/trueforge/src/controller/internalTrueForgeClient.ts index 36570085f..aeea808ac 100644 --- a/packages/trueforge/src/controller/internalTrueForgeClient.ts +++ b/packages/trueforge/src/controller/internalTrueForgeClient.ts @@ -35,4 +35,21 @@ export const internalTrueForgeClient = { async progressSandboxEnvironmentVersion(environmentVersionId: string): Promise { await client.internal.sandboxEnvironments.progress({ environmentVersionId }); }, + + async listAbandonedTurns(): Promise<{ tenantId: string; sessionId: string; turnId: string }[]> { + const response = await client.internal.turns.listAbandoned(); + return response.data.map(row => ({ + tenantId: row.tenantId, + sessionId: row.sessionId, + turnId: row.turnId, + })); + }, + + async closeAbandonedTurn(input: { tenantId: string; sessionId: string; turnId: string }): Promise { + await client.internal.turns.closeAbandoned({ + tenantId: input.tenantId, + sessionId: input.sessionId, + turnId: input.turnId, + }); + }, }; diff --git a/packages/trueforge/src/db/postgres/session-store/PostgresSessionStore.ts b/packages/trueforge/src/db/postgres/session-store/PostgresSessionStore.ts index f4316cbda..3402721dc 100644 --- a/packages/trueforge/src/db/postgres/session-store/PostgresSessionStore.ts +++ b/packages/trueforge/src/db/postgres/session-store/PostgresSessionStore.ts @@ -18,6 +18,7 @@ import type { ISessionStore, ListSessionEventsInput, ListSessionsInput, + ListStaleRunningTurnsInput, ListTurnEventsInput, ListTurnsInput, OverwriteThreadContextInput, @@ -25,6 +26,7 @@ import type { PatchSandboxInfoInput, PatchThreadCapabilityStateInput, RemoveThreadsInput, + StaleRunningTurnRef, TurnRecordWithoutSnapshot, UpdateSessionInput, UpdateTurnNonTerminalStateInput, @@ -65,6 +67,7 @@ import { createTurn as createTurnQuery, freezeAndGetTurn as freezeAndGetTurnQuery, getTurn as getTurnQuery, + listStaleRunningTurns as listStaleRunningTurnsQuery, listTurns as listTurnsQuery, updateTurnNonTerminalState as updateTurnNonTerminalStateQuery, updateTurnTerminalState as updateTurnTerminalStateQuery, @@ -171,6 +174,10 @@ export class PostgresSessionStore implements ISessionStore { + return listStaleRunningTurnsQuery(this.db, input); + } + async listTurns( input: ListTurnsInput, ): Promise<{ data: TurnRecordWithoutSnapshot[]; pagination: TokenPagination }> { diff --git a/packages/trueforge/src/db/postgres/session-store/queries/turns.ts b/packages/trueforge/src/db/postgres/session-store/queries/turns.ts index 977a49d8c..659c0d89d 100644 --- a/packages/trueforge/src/db/postgres/session-store/queries/turns.ts +++ b/packages/trueforge/src/db/postgres/session-store/queries/turns.ts @@ -9,6 +9,8 @@ import { import { assertCreateTurnThreadDelta } from '@truefoundry/trueforge-core/agent-session/store/assertCreateTurnThreadDelta'; import type { FreezeAndGetTurnInput, + ListStaleRunningTurnsInput, + StaleRunningTurnRef, TurnRecordWithoutSnapshot, UpdateTurnNonTerminalStateInput, UpdateTurnTerminalStateInput, @@ -758,6 +760,48 @@ export async function listTurns(db: Kysely, input: ListTurnsInput): Pr }; } +/** + * listStaleRunningTurns — running turns whose last event (or created_at) is before older_than. + * Oldest-first; no limit. Joins session for tenant_id (needed to load the session on close). + * Last-event aggregate is restricted to currently running turns so finished turns' events + * are not scanned. + */ +export async function listStaleRunningTurns( + db: Kysely, + input: ListStaleRunningTurnsInput, +): Promise { + const rows = await db + .selectFrom('turn as t') + .innerJoin('session as s', 's.session_id', 't.session_id') + .leftJoin( + eb => + eb + .selectFrom('session_event as e') + .innerJoin('turn as rt', join => + join.onRef('rt.session_id', '=', 'e.session_id').onRef('rt.turn_id', '=', 'e.turn_id'), + ) + .select(['e.session_id', 'e.turn_id']) + .select(eb2 => eb2.fn.max('e.created_at').as('last_event_at')) + .where(sql`rt.state->>'status' = 'running'`) + .groupBy(['e.session_id', 'e.turn_id']) + .as('e'), + join => join.onRef('e.session_id', '=', 't.session_id').onRef('e.turn_id', '=', 't.turn_id'), + ) + .select(['s.tenant_id', 't.session_id', 't.turn_id']) + .where(sql`t.state->>'status' = 'running'`) + .where(sql`COALESCE(e.last_event_at, t.created_at) < ${input.older_than}`) + .orderBy(sql`COALESCE(e.last_event_at, t.created_at)`, 'asc') + .orderBy('t.session_id', 'asc') + .orderBy('t.turn_id', 'asc') + .execute(); + + return rows.map(row => ({ + tenant_id: row.tenant_id, + session_id: row.session_id, + turn_id: row.turn_id, + })); +} + /** Atomically transitions running ↔ paused and appends turn.update. */ export async function updateTurnNonTerminalState( db: Kysely, diff --git a/packages/trueforge/src/db/sqlite/session-store/SqliteSessionStore.ts b/packages/trueforge/src/db/sqlite/session-store/SqliteSessionStore.ts index 0bd5d876e..8e65eed5f 100644 --- a/packages/trueforge/src/db/sqlite/session-store/SqliteSessionStore.ts +++ b/packages/trueforge/src/db/sqlite/session-store/SqliteSessionStore.ts @@ -18,6 +18,7 @@ import type { ISessionStore, ListSessionEventsInput, ListSessionsInput, + ListStaleRunningTurnsInput, ListTurnEventsInput, ListTurnsInput, OverwriteThreadContextInput, @@ -25,6 +26,7 @@ import type { PatchSandboxInfoInput, PatchThreadCapabilityStateInput, RemoveThreadsInput, + StaleRunningTurnRef, TurnRecordWithoutSnapshot, UpdateSessionInput, UpdateTurnNonTerminalStateInput, @@ -65,6 +67,7 @@ import { createTurn as createTurnQuery, freezeAndGetTurn as freezeAndGetTurnQuery, getTurn as getTurnQuery, + listStaleRunningTurns as listStaleRunningTurnsQuery, listTurns as listTurnsQuery, updateTurnNonTerminalState as updateTurnNonTerminalStateQuery, updateTurnTerminalState as updateTurnTerminalStateQuery, @@ -169,6 +172,10 @@ export class SqliteSessionStore implements ISessionStore { + return listStaleRunningTurnsQuery(this.db, input); + } + async listTurns( input: ListTurnsInput, ): Promise<{ data: TurnRecordWithoutSnapshot[]; pagination: TokenPagination }> { diff --git a/packages/trueforge/src/db/sqlite/session-store/queries/turns.ts b/packages/trueforge/src/db/sqlite/session-store/queries/turns.ts index 5d4ad9115..3183c6699 100644 --- a/packages/trueforge/src/db/sqlite/session-store/queries/turns.ts +++ b/packages/trueforge/src/db/sqlite/session-store/queries/turns.ts @@ -9,6 +9,8 @@ import { import { assertCreateTurnThreadDelta } from '@truefoundry/trueforge-core/agent-session/store/assertCreateTurnThreadDelta'; import type { FreezeAndGetTurnInput, + ListStaleRunningTurnsInput, + StaleRunningTurnRef, TurnRecordWithoutSnapshot, UpdateTurnNonTerminalStateInput, UpdateTurnTerminalStateInput, @@ -810,6 +812,50 @@ export async function listTurns(db: Kysely, input: ListTurnsInput): Pr }; } +/** + * listStaleRunningTurns — running turns whose last event (or created_at) is before older_than. + * Oldest-first; no limit. Timestamps are ISO strings on SQLite. + * Joins session for tenant_id (needed to load the session on close). + * Last-event aggregate is restricted to currently running turns so finished turns' events + * are not scanned. + */ +export async function listStaleRunningTurns( + db: Kysely, + input: ListStaleRunningTurnsInput, +): Promise { + const olderThanIso = input.older_than.toISOString(); + const rows = await db + .selectFrom('turn as t') + .innerJoin('session as s', 's.session_id', 't.session_id') + .leftJoin( + eb => + eb + .selectFrom('session_event as e') + .innerJoin('turn as rt', join => + join.onRef('rt.session_id', '=', 'e.session_id').onRef('rt.turn_id', '=', 'e.turn_id'), + ) + .select(['e.session_id', 'e.turn_id']) + .select(eb2 => eb2.fn.max('e.created_at').as('last_event_at')) + .where(sql`rt.state->>'status' = 'running'`) + .groupBy(['e.session_id', 'e.turn_id']) + .as('e'), + join => join.onRef('e.session_id', '=', 't.session_id').onRef('e.turn_id', '=', 't.turn_id'), + ) + .select(['s.tenant_id', 't.session_id', 't.turn_id']) + .where(sql`t.state->>'status' = 'running'`) + .where(sql`COALESCE(e.last_event_at, t.created_at) < ${olderThanIso}`) + .orderBy(sql`COALESCE(e.last_event_at, t.created_at)`, 'asc') + .orderBy('t.session_id', 'asc') + .orderBy('t.turn_id', 'asc') + .execute(); + + return rows.map(row => ({ + tenant_id: row.tenant_id, + session_id: row.session_id, + turn_id: row.turn_id, + })); +} + /** Atomically transitions running ↔ paused and appends turn.update. */ export async function updateTurnNonTerminalState( db: Kysely, diff --git a/packages/trueforge/src/routes/abandonedTurnRoutes.ts b/packages/trueforge/src/routes/abandonedTurnRoutes.ts new file mode 100644 index 000000000..09427827a --- /dev/null +++ b/packages/trueforge/src/routes/abandonedTurnRoutes.ts @@ -0,0 +1,57 @@ +/** + * Internal abandoned-turn closer routes (API-key auth). + * Mounted at /api/internal/turns. + */ +import { createRoute } from '@hono/zod-openapi'; +import { CloseAbandonedTurnRequestSchema, ListAbandonedTurnsResponseSchema } from '../schemas/abandonedTurn'; +import { RequestErrorResponseSchema } from '../schemas/errors'; +import { OpenApiTag } from './openapiTags'; + +export const listAbandonedTurnsRoute = createRoute({ + method: 'get', + path: '/abandoned', + tags: [OpenApiTag.INTERNAL], + summary: 'List abandoned running turns', + description: 'Returns all running turns whose last activity is past the stale threshold.', + 'x-fern-sdk-group-name': ['internal', 'turns'], + 'x-fern-sdk-method-name': 'list_abandoned', + 'x-excluded': true, + responses: { + 200: { + content: { 'application/json': { schema: ListAbandonedTurnsResponseSchema } }, + description: 'Stale running turns oldest-first.', + }, + 401: { + content: { 'application/json': { schema: RequestErrorResponseSchema } }, + description: 'Invalid service credential.', + }, + }, +}); + +export const closeAbandonedTurnRoute = createRoute({ + method: 'post', + path: '/close-abandoned', + tags: [OpenApiTag.INTERNAL], + summary: 'Close an abandoned running turn', + description: 'Cancels a stale running turn (turn.done with abandoned reason).', + 'x-fern-sdk-group-name': ['internal', 'turns'], + 'x-fern-sdk-method-name': 'close_abandoned', + 'x-excluded': true, + request: { + body: { + content: { 'application/json': { schema: CloseAbandonedTurnRequestSchema } }, + required: true, + }, + }, + responses: { + 204: { description: 'Closed (or already terminal / missing turn).' }, + 401: { + content: { 'application/json': { schema: RequestErrorResponseSchema } }, + description: 'Invalid service credential.', + }, + 404: { + content: { 'application/json': { schema: RequestErrorResponseSchema } }, + description: 'Session not found.', + }, + }, +}); diff --git a/packages/trueforge/src/schemas/abandonedTurn.ts b/packages/trueforge/src/schemas/abandonedTurn.ts new file mode 100644 index 000000000..c904b413e --- /dev/null +++ b/packages/trueforge/src/schemas/abandonedTurn.ts @@ -0,0 +1,29 @@ +/** + * Wire schemas for internal abandoned-turn closer endpoints. + */ +import { z } from '@hono/zod-openapi'; + +export const AbandonedTurnSchema = z + .object({ + tenant_id: z.string().min(1).describe('Tenant that owns the session.'), + session_id: z.string().min(1).describe('Session that owns the turn.'), + turn_id: z.string().min(1).describe('Running turn with stale last activity.'), + }) + .strict() + .openapi('AbandonedTurn'); + +export const ListAbandonedTurnsResponseSchema = z + .object({ + data: z.array(AbandonedTurnSchema), + }) + .strict() + .openapi('ListAbandonedTurnsResponse'); + +export const CloseAbandonedTurnRequestSchema = z + .object({ + tenant_id: z.string().min(1).describe('Tenant that owns the session.'), + session_id: z.string().min(1).describe('Session that owns the turn.'), + turn_id: z.string().min(1).describe('Turn to close.'), + }) + .strict() + .openapi('CloseAbandonedTurnRequest'); diff --git a/packages/trueforge/tests/unit/apis/abandonedTurns.test.ts b/packages/trueforge/tests/unit/apis/abandonedTurns.test.ts new file mode 100644 index 000000000..4dc510931 --- /dev/null +++ b/packages/trueforge/tests/unit/apis/abandonedTurns.test.ts @@ -0,0 +1,156 @@ +import { CancellationReason, InMemorySessionStore, Sessions } from '@truefoundry/trueforge-core/agent-session'; +import { createLogger } from 'winston'; +import { + makeAgentSpec, + makeCreateTurnInput, + makeTurnCreatedEvent, +} from '../../../../trueforge-core/tests/agent-session/testHelpers'; +import { createAbandonedTurnsRouter } from '../../../src/apis/abandonedTurns'; +import configuration from '../../../src/config'; +import { ActiveTurnRegistry } from '../../../src/runtime/activeTurns'; + +const TENANT = 'default'; +const SESSION_ID = 'abandoned-session'; +const logger = createLogger({ silent: true }); + +describe('abandoned turns internal API', () => { + async function setup() { + const sessionStore = new InMemorySessionStore(); + const sessions = new Sessions({ sessionStore }); + const activeTurns = new ActiveTurnRegistry(); + await sessionStore.createSession({ + tenant_id: TENANT, + session_id: SESSION_ID, + created_by_subject: { subject_id: 'user-1', subject_type: 'user', subject_display_name: 'user-1' }, + agent: { type: 'inline', spec: makeAgentSpec() }, + custom: null, + metadata: {}, + external_id: null, + source: null, + }); + const router = createAbandonedTurnsRouter({ + sessionStore, + sessions, + activeTurns, + logger, + }); + return { sessionStore, sessions, router }; + } + + it('lists only stale running turns and closes them as abandoned', async () => { + const { sessionStore, router } = await setup(); + const staleAt = new Date(Date.now() - configuration.ABANDONED_TURN_STALE_MS - 60_000).toISOString(); + const freshAt = new Date().toISOString(); + + await sessionStore.createTurn(makeCreateTurnInput({ sessionId: SESSION_ID, turnId: 'stale-1' })); + await sessionStore.appendToEvents({ + session_id: SESSION_ID, + turn_id: 'stale-1', + events: [{ ...makeTurnCreatedEvent('stale-1'), created_at: staleAt }], + }); + + await sessionStore.createTurn( + makeCreateTurnInput({ + sessionId: SESSION_ID, + turnId: 'fresh-1', + previousTurnId: null, + firstTurnId: 'fresh-1', + }), + ); + await sessionStore.appendToEvents({ + session_id: SESSION_ID, + turn_id: 'fresh-1', + events: [{ ...makeTurnCreatedEvent('fresh-1'), created_at: freshAt }], + }); + + const listRes = await router.request('/abandoned'); + expect(listRes.status).toBe(200); + expect(await listRes.json()).toEqual({ + data: [{ tenant_id: TENANT, session_id: SESSION_ID, turn_id: 'stale-1' }], + }); + + const closeRes = await router.request('/close-abandoned', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + tenant_id: TENANT, + session_id: SESSION_ID, + turn_id: 'stale-1', + }), + }); + expect(closeRes.status).toBe(204); + + const closed = await sessionStore.getTurn({ session_id: SESSION_ID, turn_id: 'stale-1' }); + expect(closed?.state).toMatchObject({ + status: 'cancelled', + reason: CancellationReason.Abandoned, + }); + + const fresh = await sessionStore.getTurn({ session_id: SESSION_ID, turn_id: 'fresh-1' }); + expect(fresh?.state.status).toBe('running'); + + const listAfter = await router.request('/abandoned'); + expect(listAfter.status).toBe(200); + expect(await listAfter.json()).toEqual({ data: [] }); + }); + + it('returns 404 when closing a missing session', async () => { + const { router } = await setup(); + const response = await router.request('/close-abandoned', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + tenant_id: TENANT, + session_id: 'missing-session', + turn_id: 'turn-1', + }), + }); + expect(response.status).toBe(404); + }); + + it('close is a no-op 204 when the turn is already terminal', async () => { + const { sessionStore, router } = await setup(); + const staleAt = new Date(Date.now() - configuration.ABANDONED_TURN_STALE_MS - 60_000).toISOString(); + await sessionStore.createTurn(makeCreateTurnInput({ sessionId: SESSION_ID, turnId: 'done-1' })); + await sessionStore.appendToEvents({ + session_id: SESSION_ID, + turn_id: 'done-1', + events: [{ ...makeTurnCreatedEvent('done-1'), created_at: staleAt }], + }); + await sessionStore.updateTurnTerminalState({ + session_id: SESSION_ID, + turn_id: 'done-1', + state: { + status: 'done', + output: null, + required_actions: [], + completed_at: new Date().toISOString(), + }, + turn_done_event: { + type: 'turn.done', + id: 'evt-done-1', + created_at: new Date().toISOString(), + state: { + status: 'done', + output: null, + required_actions: [], + completed_at: new Date().toISOString(), + }, + thread_id: null, + }, + }); + + const response = await router.request('/close-abandoned', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + tenant_id: TENANT, + session_id: SESSION_ID, + turn_id: 'done-1', + }), + }); + expect(response.status).toBe(204); + const turn = await sessionStore.getTurn({ session_id: SESSION_ID, turn_id: 'done-1' }); + expect(turn?.state.status).toBe('done'); + }); +}); diff --git a/packages/trueforge/tests/unit/controller/abandonedTurnCloser.test.ts b/packages/trueforge/tests/unit/controller/abandonedTurnCloser.test.ts new file mode 100644 index 000000000..ebebf5a9a --- /dev/null +++ b/packages/trueforge/tests/unit/controller/abandonedTurnCloser.test.ts @@ -0,0 +1,57 @@ +import { TrueForgeApi } from '@truefoundry/trueforge-sdk'; +import configuration from '../../../src/config'; +import { abandonedTurnCloserLoop, dispatchAbandonedTurns } from '../../../src/controller/abandonedTurnCloser'; +import { internalTrueForgeClient } from '../../../src/controller/internalTrueForgeClient'; + +function fakeLogger() { + return { error: jest.fn(), debug: jest.fn(), info: jest.fn(), warn: jest.fn() }; +} + +function abandoned(n: number): { tenantId: string; sessionId: string; turnId: string } { + return { tenantId: 't1', sessionId: `s-${String(n)}`, turnId: `turn-${String(n)}` }; +} + +describe('abandonedTurnCloserLoop', () => { + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('closes each abandoned turn from the internal API up to the batch limit', async () => { + const batch = configuration.ABANDONED_TURN_CLOSE_BATCH; + const all = Array.from({ length: batch + 5 }, (_, i) => abandoned(i)); + jest.spyOn(internalTrueForgeClient, 'listAbandonedTurns').mockResolvedValue(all); + const close = jest.spyOn(internalTrueForgeClient, 'closeAbandonedTurn').mockResolvedValue(undefined); + + const logger = fakeLogger(); + const loop = abandonedTurnCloserLoop({ logger: logger as never }); + await loop.tick(new AbortController().signal); + + expect(close).toHaveBeenCalledTimes(batch); + expect(close).toHaveBeenNthCalledWith(1, abandoned(0)); + expect(close).toHaveBeenNthCalledWith(batch, abandoned(batch - 1)); + }); + + it('skips NotFoundError and continues the tick', async () => { + jest.spyOn(internalTrueForgeClient, 'listAbandonedTurns').mockResolvedValue([abandoned(1), abandoned(2)]); + const close = jest.spyOn(internalTrueForgeClient, 'closeAbandonedTurn').mockImplementation(async input => { + if (input.turnId === 'turn-1') { + throw new TrueForgeApi.NotFoundError({ error: { message: 'not found' } }); + } + }); + const logger = fakeLogger(); + + await dispatchAbandonedTurns({ logger: logger as never }); + + expect(close).toHaveBeenCalledTimes(2); + expect(logger.warn).toHaveBeenCalled(); + }); + + it('does not tick when aborted', async () => { + const list = jest.spyOn(internalTrueForgeClient, 'listAbandonedTurns'); + const controller = new AbortController(); + controller.abort(); + const loop = abandonedTurnCloserLoop({ logger: fakeLogger() as never }); + await loop.tick(controller.signal); + expect(list).not.toHaveBeenCalled(); + }); +}); diff --git a/python/trueforge_sdk/reference.md b/python/trueforge_sdk/reference.md index 46fef12c4..f21653c56 100644 --- a/python/trueforge_sdk/reference.md +++ b/python/trueforge_sdk/reference.md @@ -4379,6 +4379,159 @@ client.internal.sessions.get_or_create_by_external_id( + + + + +## Internal Turns +
client.internal.turns.list_abandoned() -> ListAbandonedTurnsResponse +
+
+ +#### 📝 Description + +
+
+ +
+
+ +Returns all running turns whose last activity is past the stale threshold. +
+
+
+
+ +#### 🔌 Usage + +
+
+ +
+
+ +```python +from trueforge_sdk import TrueForge + +client = TrueForge( + token="", + base_url="https://yourhost.com/path/to/api", +) + +client.internal.turns.list_abandoned() + +``` +
+
+
+
+ +#### ⚙️ Parameters + +
+
+ +
+
+ +**request_options:** `typing.Optional[RequestOptions]` — Request-specific configuration. + +
+
+
+
+ + +
+
+
+ +
client.internal.turns.close_abandoned(...) +
+
+ +#### 📝 Description + +
+
+ +
+
+ +Cancels a stale running turn (turn.done with abandoned reason). +
+
+
+
+ +#### 🔌 Usage + +
+
+ +
+
+ +```python +from trueforge_sdk import TrueForge + +client = TrueForge( + token="", + base_url="https://yourhost.com/path/to/api", +) + +client.internal.turns.close_abandoned( + session_id="session_id", + tenant_id="tenant_id", + turn_id="turn_id", +) + +``` +
+
+
+
+ +#### ⚙️ Parameters + +
+
+ +
+
+ +**session_id:** `str` — Session that owns the turn. + +
+
+ +
+
+ +**tenant_id:** `str` — Tenant that owns the session. + +
+
+ +
+
+ +**turn_id:** `str` — Turn to close. + +
+
+ +
+
+ +**request_options:** `typing.Optional[RequestOptions]` — Request-specific configuration. + +
+
+
+
+ +
diff --git a/python/trueforge_sdk/src/trueforge_sdk/__init__.py b/python/trueforge_sdk/src/trueforge_sdk/__init__.py index c4ccacf02..3b1f5c430 100644 --- a/python/trueforge_sdk/src/trueforge_sdk/__init__.py +++ b/python/trueforge_sdk/src/trueforge_sdk/__init__.py @@ -7,6 +7,7 @@ if typing.TYPE_CHECKING: from .types import ( + AbandonedTurn, ActionRequired, ActionRequiredEvent, Agent, @@ -103,6 +104,7 @@ InitialUserMessage, InputTokensCompactionTrigger, LargeToolResponseConfig, + ListAbandonedTurnsResponse, ListAgentsResponse, ListAvailableMcpServersResponse, ListAvailableModelsResponse, @@ -309,6 +311,7 @@ from .client import AsyncTrueForge, TrueForge from .version import __version__ _dynamic_imports: typing.Dict[str, str] = { + "AbandonedTurn": ".types", "ActionRequired": ".types", "ActionRequiredEvent": ".types", "Agent": ".types", @@ -416,6 +419,7 @@ "InputTokensCompactionTrigger": ".types", "InternalServerError": ".errors", "LargeToolResponseConfig": ".types", + "ListAbandonedTurnsResponse": ".types", "ListAgentsResponse": ".types", "ListAvailableMcpServersResponse": ".types", "ListAvailableModelsResponse": ".types", @@ -632,6 +636,7 @@ def __dir__(): __all__ = [ + "AbandonedTurn", "ActionRequired", "ActionRequiredEvent", "Agent", @@ -739,6 +744,7 @@ def __dir__(): "InputTokensCompactionTrigger", "InternalServerError", "LargeToolResponseConfig", + "ListAbandonedTurnsResponse", "ListAgentsResponse", "ListAvailableMcpServersResponse", "ListAvailableModelsResponse", diff --git a/python/trueforge_sdk/src/trueforge_sdk/internal/__init__.py b/python/trueforge_sdk/src/trueforge_sdk/internal/__init__.py index e74a16a7e..9c236edaf 100644 --- a/python/trueforge_sdk/src/trueforge_sdk/internal/__init__.py +++ b/python/trueforge_sdk/src/trueforge_sdk/internal/__init__.py @@ -6,13 +6,14 @@ from importlib import import_module if typing.TYPE_CHECKING: - from . import agents, metrics, sandbox_environments, schedules, sessions + from . import agents, metrics, sandbox_environments, schedules, sessions, turns _dynamic_imports: typing.Dict[str, str] = { "agents": ".agents", "metrics": ".metrics", "sandbox_environments": ".sandbox_environments", "schedules": ".schedules", "sessions": ".sessions", + "turns": ".turns", } @@ -37,4 +38,4 @@ def __dir__(): return sorted(lazy_attrs) -__all__ = ["agents", "metrics", "sandbox_environments", "schedules", "sessions"] +__all__ = ["agents", "metrics", "sandbox_environments", "schedules", "sessions", "turns"] diff --git a/python/trueforge_sdk/src/trueforge_sdk/internal/client.py b/python/trueforge_sdk/src/trueforge_sdk/internal/client.py index ee9183905..ebc870b9d 100644 --- a/python/trueforge_sdk/src/trueforge_sdk/internal/client.py +++ b/python/trueforge_sdk/src/trueforge_sdk/internal/client.py @@ -16,6 +16,7 @@ from .sandbox_environments.client import AsyncSandboxEnvironmentsClient, SandboxEnvironmentsClient from .schedules.client import AsyncSchedulesClient, SchedulesClient from .sessions.client import AsyncSessionsClient, SessionsClient + from .turns.client import AsyncTurnsClient, TurnsClient # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) @@ -28,6 +29,7 @@ def __init__(self, *, client_wrapper: SyncClientWrapper): self._sandbox_environments: typing.Optional[SandboxEnvironmentsClient] = None self._schedules: typing.Optional[SchedulesClient] = None self._sessions: typing.Optional[SessionsClient] = None + self._turns: typing.Optional[TurnsClient] = None self._agents: typing.Optional[AgentsClient] = None @property @@ -116,6 +118,14 @@ def sessions(self): self._sessions = SessionsClient(client_wrapper=self._client_wrapper) return self._sessions + @property + def turns(self): + if self._turns is None: + from .turns.client import TurnsClient # noqa: E402 + + self._turns = TurnsClient(client_wrapper=self._client_wrapper) + return self._turns + @property def agents(self): if self._agents is None: @@ -133,6 +143,7 @@ def __init__(self, *, client_wrapper: AsyncClientWrapper): self._sandbox_environments: typing.Optional[AsyncSandboxEnvironmentsClient] = None self._schedules: typing.Optional[AsyncSchedulesClient] = None self._sessions: typing.Optional[AsyncSessionsClient] = None + self._turns: typing.Optional[AsyncTurnsClient] = None self._agents: typing.Optional[AsyncAgentsClient] = None @property @@ -229,6 +240,14 @@ def sessions(self): self._sessions = AsyncSessionsClient(client_wrapper=self._client_wrapper) return self._sessions + @property + def turns(self): + if self._turns is None: + from .turns.client import AsyncTurnsClient # noqa: E402 + + self._turns = AsyncTurnsClient(client_wrapper=self._client_wrapper) + return self._turns + @property def agents(self): if self._agents is None: diff --git a/python/trueforge_sdk/src/trueforge_sdk/internal/turns/__init__.py b/python/trueforge_sdk/src/trueforge_sdk/internal/turns/__init__.py new file mode 100644 index 000000000..5cde0202d --- /dev/null +++ b/python/trueforge_sdk/src/trueforge_sdk/internal/turns/__init__.py @@ -0,0 +1,4 @@ +# This file was auto-generated by Fern from our API Definition. + +# isort: skip_file + diff --git a/python/trueforge_sdk/src/trueforge_sdk/internal/turns/client.py b/python/trueforge_sdk/src/trueforge_sdk/internal/turns/client.py new file mode 100644 index 000000000..800148c56 --- /dev/null +++ b/python/trueforge_sdk/src/trueforge_sdk/internal/turns/client.py @@ -0,0 +1,201 @@ +# This file was auto-generated by Fern from our API Definition. + +import typing + +from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper +from ...core.request_options import RequestOptions +from ...types.list_abandoned_turns_response import ListAbandonedTurnsResponse +from .raw_client import AsyncRawTurnsClient, RawTurnsClient + +# this is used as the default value for optional parameters +OMIT = typing.cast(typing.Any, ...) + + +class TurnsClient: + def __init__(self, *, client_wrapper: SyncClientWrapper): + self._raw_client = RawTurnsClient(client_wrapper=client_wrapper) + + @property + def with_raw_response(self) -> RawTurnsClient: + """ + Retrieves a raw implementation of this client that returns raw responses. + + Returns + ------- + RawTurnsClient + """ + return self._raw_client + + def list_abandoned(self, *, request_options: typing.Optional[RequestOptions] = None) -> ListAbandonedTurnsResponse: + """ + Returns all running turns whose last activity is past the stale threshold. + + Parameters + ---------- + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + ListAbandonedTurnsResponse + Stale running turns oldest-first. + + Examples + -------- + from trueforge_sdk import TrueForge + + client = TrueForge( + token="YOUR_TOKEN", + base_url="https://yourhost.com/path/to/api", + ) + client.internal.turns.list_abandoned() + """ + _response = self._raw_client.list_abandoned(request_options=request_options) + return _response.data + + def close_abandoned( + self, *, session_id: str, tenant_id: str, turn_id: str, request_options: typing.Optional[RequestOptions] = None + ) -> None: + """ + Cancels a stale running turn (turn.done with abandoned reason). + + Parameters + ---------- + session_id : str + Session that owns the turn. + + tenant_id : str + Tenant that owns the session. + + turn_id : str + Turn to close. + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + None + + Examples + -------- + from trueforge_sdk import TrueForge + + client = TrueForge( + token="YOUR_TOKEN", + base_url="https://yourhost.com/path/to/api", + ) + client.internal.turns.close_abandoned( + session_id="session_id", + tenant_id="tenant_id", + turn_id="turn_id", + ) + """ + _response = self._raw_client.close_abandoned( + session_id=session_id, tenant_id=tenant_id, turn_id=turn_id, request_options=request_options + ) + return _response.data + + +class AsyncTurnsClient: + def __init__(self, *, client_wrapper: AsyncClientWrapper): + self._raw_client = AsyncRawTurnsClient(client_wrapper=client_wrapper) + + @property + def with_raw_response(self) -> AsyncRawTurnsClient: + """ + Retrieves a raw implementation of this client that returns raw responses. + + Returns + ------- + AsyncRawTurnsClient + """ + return self._raw_client + + async def list_abandoned( + self, *, request_options: typing.Optional[RequestOptions] = None + ) -> ListAbandonedTurnsResponse: + """ + Returns all running turns whose last activity is past the stale threshold. + + Parameters + ---------- + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + ListAbandonedTurnsResponse + Stale running turns oldest-first. + + Examples + -------- + import asyncio + + from trueforge_sdk import AsyncTrueForge + + client = AsyncTrueForge( + token="YOUR_TOKEN", + base_url="https://yourhost.com/path/to/api", + ) + + + async def main() -> None: + await client.internal.turns.list_abandoned() + + + asyncio.run(main()) + """ + _response = await self._raw_client.list_abandoned(request_options=request_options) + return _response.data + + async def close_abandoned( + self, *, session_id: str, tenant_id: str, turn_id: str, request_options: typing.Optional[RequestOptions] = None + ) -> None: + """ + Cancels a stale running turn (turn.done with abandoned reason). + + Parameters + ---------- + session_id : str + Session that owns the turn. + + tenant_id : str + Tenant that owns the session. + + turn_id : str + Turn to close. + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + None + + Examples + -------- + import asyncio + + from trueforge_sdk import AsyncTrueForge + + client = AsyncTrueForge( + token="YOUR_TOKEN", + base_url="https://yourhost.com/path/to/api", + ) + + + async def main() -> None: + await client.internal.turns.close_abandoned( + session_id="session_id", + tenant_id="tenant_id", + turn_id="turn_id", + ) + + + asyncio.run(main()) + """ + _response = await self._raw_client.close_abandoned( + session_id=session_id, tenant_id=tenant_id, turn_id=turn_id, request_options=request_options + ) + return _response.data diff --git a/python/trueforge_sdk/src/trueforge_sdk/internal/turns/raw_client.py b/python/trueforge_sdk/src/trueforge_sdk/internal/turns/raw_client.py new file mode 100644 index 000000000..9d065cf29 --- /dev/null +++ b/python/trueforge_sdk/src/trueforge_sdk/internal/turns/raw_client.py @@ -0,0 +1,275 @@ +# This file was auto-generated by Fern from our API Definition. + +import typing +from json.decoder import JSONDecodeError + +from ...core.api_error import ApiError +from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper +from ...core.http_response import AsyncHttpResponse, HttpResponse +from ...core.parse_error import ParsingError +from ...core.request_options import RequestOptions +from ...core.unchecked_base_model import construct_type +from ...errors.not_found_error import NotFoundError +from ...errors.unauthorized_error import UnauthorizedError +from ...types.list_abandoned_turns_response import ListAbandonedTurnsResponse +from ...types.request_error_response import RequestErrorResponse +from pydantic import ValidationError + +# this is used as the default value for optional parameters +OMIT = typing.cast(typing.Any, ...) + + +class RawTurnsClient: + def __init__(self, *, client_wrapper: SyncClientWrapper): + self._client_wrapper = client_wrapper + + def list_abandoned( + self, *, request_options: typing.Optional[RequestOptions] = None + ) -> HttpResponse[ListAbandonedTurnsResponse]: + """ + Returns all running turns whose last activity is past the stale threshold. + + Parameters + ---------- + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + HttpResponse[ListAbandonedTurnsResponse] + Stale running turns oldest-first. + """ + _response = self._client_wrapper.httpx_client.request( + "api/internal/turns/abandoned", + method="GET", + request_options=request_options, + ) + try: + if 200 <= _response.status_code < 300: + _data = typing.cast( + ListAbandonedTurnsResponse, + construct_type( + type_=ListAbandonedTurnsResponse, # type: ignore + object_=_response.json(), + ), + ) + return HttpResponse(response=_response, data=_data) + if _response.status_code == 401: + raise UnauthorizedError( + headers=dict(_response.headers), + body=typing.cast( + RequestErrorResponse, + construct_type( + type_=RequestErrorResponse, # type: ignore + object_=_response.json(), + ), + ), + ) + _response_json = _response.json() + except JSONDecodeError: + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) + except ValidationError as e: + raise ParsingError( + status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e + ) + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + + def close_abandoned( + self, *, session_id: str, tenant_id: str, turn_id: str, request_options: typing.Optional[RequestOptions] = None + ) -> HttpResponse[None]: + """ + Cancels a stale running turn (turn.done with abandoned reason). + + Parameters + ---------- + session_id : str + Session that owns the turn. + + tenant_id : str + Tenant that owns the session. + + turn_id : str + Turn to close. + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + HttpResponse[None] + """ + _response = self._client_wrapper.httpx_client.request( + "api/internal/turns/close-abandoned", + method="POST", + json={ + "session_id": session_id, + "tenant_id": tenant_id, + "turn_id": turn_id, + }, + headers={ + "content-type": "application/json", + }, + request_options=request_options, + omit=OMIT, + ) + try: + if 200 <= _response.status_code < 300: + return HttpResponse(response=_response, data=None) + if _response.status_code == 401: + raise UnauthorizedError( + headers=dict(_response.headers), + body=typing.cast( + RequestErrorResponse, + construct_type( + type_=RequestErrorResponse, # type: ignore + object_=_response.json(), + ), + ), + ) + if _response.status_code == 404: + raise NotFoundError( + headers=dict(_response.headers), + body=typing.cast( + RequestErrorResponse, + construct_type( + type_=RequestErrorResponse, # type: ignore + object_=_response.json(), + ), + ), + ) + _response_json = _response.json() + except JSONDecodeError: + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) + except ValidationError as e: + raise ParsingError( + status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e + ) + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + + +class AsyncRawTurnsClient: + def __init__(self, *, client_wrapper: AsyncClientWrapper): + self._client_wrapper = client_wrapper + + async def list_abandoned( + self, *, request_options: typing.Optional[RequestOptions] = None + ) -> AsyncHttpResponse[ListAbandonedTurnsResponse]: + """ + Returns all running turns whose last activity is past the stale threshold. + + Parameters + ---------- + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + AsyncHttpResponse[ListAbandonedTurnsResponse] + Stale running turns oldest-first. + """ + _response = await self._client_wrapper.httpx_client.request( + "api/internal/turns/abandoned", + method="GET", + request_options=request_options, + ) + try: + if 200 <= _response.status_code < 300: + _data = typing.cast( + ListAbandonedTurnsResponse, + construct_type( + type_=ListAbandonedTurnsResponse, # type: ignore + object_=_response.json(), + ), + ) + return AsyncHttpResponse(response=_response, data=_data) + if _response.status_code == 401: + raise UnauthorizedError( + headers=dict(_response.headers), + body=typing.cast( + RequestErrorResponse, + construct_type( + type_=RequestErrorResponse, # type: ignore + object_=_response.json(), + ), + ), + ) + _response_json = _response.json() + except JSONDecodeError: + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) + except ValidationError as e: + raise ParsingError( + status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e + ) + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + + async def close_abandoned( + self, *, session_id: str, tenant_id: str, turn_id: str, request_options: typing.Optional[RequestOptions] = None + ) -> AsyncHttpResponse[None]: + """ + Cancels a stale running turn (turn.done with abandoned reason). + + Parameters + ---------- + session_id : str + Session that owns the turn. + + tenant_id : str + Tenant that owns the session. + + turn_id : str + Turn to close. + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + AsyncHttpResponse[None] + """ + _response = await self._client_wrapper.httpx_client.request( + "api/internal/turns/close-abandoned", + method="POST", + json={ + "session_id": session_id, + "tenant_id": tenant_id, + "turn_id": turn_id, + }, + headers={ + "content-type": "application/json", + }, + request_options=request_options, + omit=OMIT, + ) + try: + if 200 <= _response.status_code < 300: + return AsyncHttpResponse(response=_response, data=None) + if _response.status_code == 401: + raise UnauthorizedError( + headers=dict(_response.headers), + body=typing.cast( + RequestErrorResponse, + construct_type( + type_=RequestErrorResponse, # type: ignore + object_=_response.json(), + ), + ), + ) + if _response.status_code == 404: + raise NotFoundError( + headers=dict(_response.headers), + body=typing.cast( + RequestErrorResponse, + construct_type( + type_=RequestErrorResponse, # type: ignore + object_=_response.json(), + ), + ), + ) + _response_json = _response.json() + except JSONDecodeError: + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) + except ValidationError as e: + raise ParsingError( + status_code=_response.status_code, headers=dict(_response.headers), body=_response.json(), cause=e + ) + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) diff --git a/python/trueforge_sdk/src/trueforge_sdk/types/__init__.py b/python/trueforge_sdk/src/trueforge_sdk/types/__init__.py index 6a92da596..2b3b2f665 100644 --- a/python/trueforge_sdk/src/trueforge_sdk/types/__init__.py +++ b/python/trueforge_sdk/src/trueforge_sdk/types/__init__.py @@ -6,6 +6,7 @@ from importlib import import_module if typing.TYPE_CHECKING: + from .abandoned_turn import AbandonedTurn from .action_required import ActionRequired from .action_required_event import ActionRequiredEvent from .agent import Agent @@ -102,6 +103,7 @@ from .initial_user_message import InitialUserMessage from .input_tokens_compaction_trigger import InputTokensCompactionTrigger from .large_tool_response_config import LargeToolResponseConfig + from .list_abandoned_turns_response import ListAbandonedTurnsResponse from .list_agents_response import ListAgentsResponse from .list_available_mcp_servers_response import ListAvailableMcpServersResponse from .list_available_models_response import ListAvailableModelsResponse @@ -276,6 +278,7 @@ from .web_search_provider_manifest import WebSearchProviderManifest from .zai_model_provider import ZaiModelProvider _dynamic_imports: typing.Dict[str, str] = { + "AbandonedTurn": ".abandoned_turn", "ActionRequired": ".action_required", "ActionRequiredEvent": ".action_required_event", "Agent": ".agent", @@ -372,6 +375,7 @@ "InitialUserMessage": ".initial_user_message", "InputTokensCompactionTrigger": ".input_tokens_compaction_trigger", "LargeToolResponseConfig": ".large_tool_response_config", + "ListAbandonedTurnsResponse": ".list_abandoned_turns_response", "ListAgentsResponse": ".list_agents_response", "ListAvailableMcpServersResponse": ".list_available_mcp_servers_response", "ListAvailableModelsResponse": ".list_available_models_response", @@ -570,6 +574,7 @@ def __dir__(): __all__ = [ + "AbandonedTurn", "ActionRequired", "ActionRequiredEvent", "Agent", @@ -666,6 +671,7 @@ def __dir__(): "InitialUserMessage", "InputTokensCompactionTrigger", "LargeToolResponseConfig", + "ListAbandonedTurnsResponse", "ListAgentsResponse", "ListAvailableMcpServersResponse", "ListAvailableModelsResponse", diff --git a/python/trueforge_sdk/src/trueforge_sdk/types/abandoned_turn.py b/python/trueforge_sdk/src/trueforge_sdk/types/abandoned_turn.py new file mode 100644 index 000000000..1a90a1973 --- /dev/null +++ b/python/trueforge_sdk/src/trueforge_sdk/types/abandoned_turn.py @@ -0,0 +1,32 @@ +# This file was auto-generated by Fern from our API Definition. + +import typing + +import pydantic +from ..core.pydantic_utilities import IS_PYDANTIC_V2 +from ..core.unchecked_base_model import UncheckedBaseModel + + +class AbandonedTurn(UncheckedBaseModel): + session_id: str = pydantic.Field() + """ + Session that owns the turn. + """ + + tenant_id: str = pydantic.Field() + """ + Tenant that owns the session. + """ + + turn_id: str = pydantic.Field() + """ + Running turn with stale last activity. + """ + + if IS_PYDANTIC_V2: + model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict(extra="allow") # type: ignore # Pydantic v2 + else: + + class Config: + smart_union = True + extra = pydantic.Extra.allow diff --git a/python/trueforge_sdk/src/trueforge_sdk/types/list_abandoned_turns_response.py b/python/trueforge_sdk/src/trueforge_sdk/types/list_abandoned_turns_response.py new file mode 100644 index 000000000..eaf952572 --- /dev/null +++ b/python/trueforge_sdk/src/trueforge_sdk/types/list_abandoned_turns_response.py @@ -0,0 +1,20 @@ +# This file was auto-generated by Fern from our API Definition. + +import typing + +import pydantic +from ..core.pydantic_utilities import IS_PYDANTIC_V2 +from ..core.unchecked_base_model import UncheckedBaseModel +from .abandoned_turn import AbandonedTurn + + +class ListAbandonedTurnsResponse(UncheckedBaseModel): + data: typing.List[AbandonedTurn] + + if IS_PYDANTIC_V2: + model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict(extra="allow") # type: ignore # Pydantic v2 + else: + + class Config: + smart_union = True + extra = pydantic.Extra.allow