feat: add table for session inbound events

.changeset/session-inbound-events.md

@@ -0,0 +1,6 @@
+---
+"@truefoundry/trueforge-core": patch
+"@truefoundry/trueforge": patch
+---
+
+Add `session_inbound_events` store API for durable tip HITL send-event inbox (insert / list unconsumed / mark consumed), with Postgres and SQLite migrations.

packages/trueforge-core/src/agent-session/index.ts

@@ -21,6 +21,9 @@ export {
 } from './schemas/turn';
 export type { TerminalTurnState, Turn, TurnInputItem, TurnMetrics, TurnState } from './schemas/turn';
 
+export { SessionInboundEventItemSchema } from './schemas/sendEvent';
+export type { SessionInboundEventItem } from './schemas/sendEvent';
+
 export {
   SessionMetadataSchema,
   SessionMetricsSchema,
@@ -80,16 +83,20 @@ export type {
   GetSessionInput,
   GetTurnInput,
   ISessionStore,
+  InsertSessionInboundEventsInput,
   ListSessionEventsInput,
   ListSessionsInput,
   ListTurnEventsInput,
   ListTurnsInput,
+  ListUnconsumedSessionInboundEventsInput,
+  MarkSessionInboundEventsConsumedInput,
   NewThreadInit,
   OverwriteThreadContextInput,
   PatchMCPServersInput,
   PatchSandboxInfoInput,
   PatchThreadCapabilityStateInput,
   RemoveThreadsInput,
+  SessionInboundEventRecord,
   TurnContextAppend,
   TurnRecordWithoutSnapshot,
   UpdateSessionInput,
@@ -100,6 +107,7 @@ export {
   PreviousTurnRunningError,
   SessionAlreadyExistsError,
   SessionExternalIdConflictError,
+  SessionInboundEventAlreadyExistsError,
   SessionNotFoundError,
   SessionStoreConflictError,
   SessionStoreInvariantError,

packages/trueforge-core/src/agent-session/schemas/sendEvent.ts

@@ -0,0 +1,17 @@
+/**
+ * Inbound send-event payloads for tip HITL (client → harness), distinct from the
+ * stream log ({@link PersistedTurnEvent} / session_event).
+ *
+ * Public send is session-scoped (`POST …/sessions/{id}/events`) with required
+ * body `turn_id` (one batch → one tip) plus `SessionInboundEventItem`s; rows stamp that
+ * tip id. v1 union is tip-only; approval policies may relax `turn_id` later.
+ * `user.message` stays on createTurn / steer.
+ */
+import { z } from '@hono/zod-openapi';
+import { UserToolApprovalMessageSchema, UserToolResponseMessageSchema } from '../../core/events/schema';
+
+export const SessionInboundEventItemSchema = z
+  .discriminatedUnion('type', [UserToolApprovalMessageSchema, UserToolResponseMessageSchema])
+  .openapi('SessionInboundEventItem');
+
+export type SessionInboundEventItem = z.infer<typeof SessionInboundEventItemSchema>;

packages/trueforge-core/src/agent-session/store/ISessionStore.ts

@@ -11,6 +11,7 @@
 import type { TurnRecord } from '../models/TurnRecord';
 import type { PersistedTurnEvent, SessionEventItem } from '../schemas/events';
 import type { TokenPagination } from '../schemas/pagination';
+import type { SessionInboundEventItem } from '../schemas/sendEvent';
 import type { SessionMetadata } from '../schemas/session';
 import type { CancellationReason, TerminalTurnState } from '../schemas/turn';
 
@@ -170,6 +171,52 @@
   events: PersistedTurnEvent[];
 }
 
+/** One durable inbound send-event row (tip HITL and/or session-scoped). */
+export interface SessionInboundEventRecord {
+  event_id: string;
+  /** Tip id when tip-scoped; null for session-only (e.g. future policies). */
+  turn_id: string | null;
+  /** Validated {@link SessionInboundEventItem} body (widens when policy lands). */
+  payload: SessionInboundEventItem;
+  /** ISO-8601; copied from insert input. Ordering uses `event_id`. */
+  created_at: string;
+}
+
+export interface InsertSessionInboundEventsInput {
+  session_id: string;
+  /**
+   * Tip that receives this batch (v1 required). One send = one tip; stamp every
+   * row with this id. Relax to optional/null when session-scoped policies land.
+   */
+  turn_id: string;
+  /**
+   * Caller mints `event_id` (monotonic ULID) — same contract as session_event.
+   * Empty array is a no-op.
+   */
+  events: Array<{
+    event_id: string;
+    payload: SessionInboundEventItem;
+    created_at: string;
+  }>;
+}
+
+export interface ListUnconsumedSessionInboundEventsInput {
+  session_id: string;
+  /**
+   * Three-way filter — pass the key explicitly (do not omit):
+   * - `undefined` — all unconsumed for the session
+   * - `string` — unconsumed for that turn only
+   * - `null` — session-scoped rows only (`turn_id` IS NULL; empty until
+   *   policies allow null inserts)
+   */
+  turn_id: string | null | undefined;
+}
+
+export interface MarkSessionInboundEventsConsumedInput {
+  session_id: string;
+  event_ids: string[];
+}
+
 export interface AddThreadsInput {
   session_id: string;
   turn_id: string;
@@ -360,6 +407,31 @@
    */
   appendToEvents(input: AppendToEventsInput): Promise<void>;
 
+  /**
+   * Durable inbound send-event inbox for the session. Column stays nullable for later
+   * session-scoped policies. Tip must be non-terminal (v1: `running`; `paused`
+   * when that status lands) — terminal tip → {@link TurnNotRunningError}.
+   * Missing session → {@link SessionNotFoundError}; unknown turn →
+   * {@link TurnNotFoundError}. Duplicate `event_id` →
+   * {@link SessionInboundEventAlreadyExistsError}.
+   */
+  insertSessionInboundEvents(input: InsertSessionInboundEventsInput): Promise<void>;
+
+  /**
+   * Unconsumed inbox rows, ordered by monotonic `event_id` ascending.
+   * See {@link ListUnconsumedSessionInboundEventsInput.turn_id} for filtering.
+   * Missing session → {@link SessionNotFoundError}.
+   */
+  listUnconsumedSessionInboundEvents(
+    input: ListUnconsumedSessionInboundEventsInput,
+  ): Promise<SessionInboundEventRecord[]>;
+
+  /**
+   * Marks inbox rows consumed. Already-consumed or unknown ids are ignored.
+   * Empty `event_ids` is a no-op. Missing session → {@link SessionNotFoundError}.
+   */
+  markSessionInboundEventsConsumed(input: MarkSessionInboundEventsConsumedInput): Promise<void>;
+
   /** Adds thread snapshots to the turn (sub-agent spawns). */
   addThreads(input: AddThreadsInput): Promise<void>;
 

packages/trueforge-core/src/agent-session/store/InMemorySessionStore.ts

@@ -4,6 +4,7 @@ import type { SessionRecord } from '../models/SessionRecord';
 import type { TurnRecord, TurnSnapshot } from '../models/TurnRecord';
 import type { PersistedTurnEvent, SessionEventItem } from '../schemas/events';
 import type { TokenPagination } from '../schemas/pagination';
+import type { SessionInboundEventItem } from '../schemas/sendEvent';
 import type { TerminalTurnState } from '../schemas/turn';
 import { assertCreateTurnThreadDelta } from './assertCreateTurnThreadDelta';
 import type {
@@ -18,17 +19,21 @@ import type {
   GetSessionByExternalIdInput,
   GetSessionInput,
   GetTurnInput,
+  InsertSessionInboundEventsInput,
   ISessionStore,
   ListSessionEventsInput,
   ListSessionsInput,
   ListTurnEventsInput,
   ListTurnsInput,
+  ListUnconsumedSessionInboundEventsInput,
+  MarkSessionInboundEventsConsumedInput,
   NewThreadInit,
   OverwriteThreadContextInput,
   PatchMCPServersInput,
   PatchSandboxInfoInput,
   PatchThreadCapabilityStateInput,
   RemoveThreadsInput,
+  SessionInboundEventRecord,
   TurnContextAppend,
   TurnRecordWithoutSnapshot,
   UpdateSessionInput,
@@ -45,6 +50,7 @@ import {
   PreviousTurnRunningError,
   SessionAlreadyExistsError,
   SessionExternalIdConflictError,
+  SessionInboundEventAlreadyExistsError,
   SessionNotFoundError,
   SessionStoreInvariantError,
   TurnAlreadyExistsError,
@@ -56,6 +62,14 @@ import {
 
 type StoredEvent = PersistedTurnEvent;
 
+interface StoredInboundEvent {
+  event_id: string;
+  turn_id: string | null;
+  payload: SessionInboundEventItem;
+  created_at: string;
+  consumed: boolean;
+}
+
 interface StoredSession<TSessionCustom extends object> {
   record: SessionRecord<TSessionCustom>;
   turnIds: string[];
@@ -170,6 +184,8 @@ export class InMemorySessionStore<
   private readonly sessions = new Map<string, StoredSession<TSessionCustom>>();
   private readonly turns = new Map<string, TurnRecord<TTurnCustom>>();
   private readonly events = new Map<string, StoredEvent[]>();
+  /** session_id → inbound send-event inbox */
+  private readonly inboundEvents = new Map<string, StoredInboundEvent[]>();
 
   async createSession(input: CreateSessionInput<TSessionCustom>): Promise<void> {
     const key = sessionKey(input.session_id);
@@ -219,6 +235,7 @@ export class InMemorySessionStore<
       this.turns.delete(tKey);
       this.events.delete(tKey);
     }
+    this.inboundEvents.delete(sessionKey(input.session_id));
     this.sessions.delete(sKey);
   }
 
@@ -485,6 +502,81 @@ export class InMemorySessionStore<
     return;
   }
 
+  async insertSessionInboundEvents(input: InsertSessionInboundEventsInput): Promise<void> {
+    if (input.events.length === 0) {
+      return;
+    }
+    this.requireSession(input.session_id);
+    this.requireRunningTurn(input.session_id, input.turn_id);
+    const sKey = sessionKey(input.session_id);
+    let list = this.inboundEvents.get(sKey);
+    if (!list) {
+      list = [];
+      this.inboundEvents.set(sKey, list);
+    }
+    const existing = new Set(list.map(row => row.event_id));
+    for (const event of input.events) {
+      if (existing.has(event.event_id)) {
+        throw new SessionInboundEventAlreadyExistsError(input.session_id, event.event_id);
+      }
+      existing.add(event.event_id);
+    }
+    for (const event of input.events) {
+      list.push({
+        event_id: event.event_id,
+        turn_id: input.turn_id,
+        payload: deepCopy(event.payload),
+        created_at: event.created_at,
+        consumed: false,
+      });
+    }
+  }
+
+  async listUnconsumedSessionInboundEvents(
+    input: ListUnconsumedSessionInboundEventsInput,
+  ): Promise<SessionInboundEventRecord[]> {
+    this.requireSession(input.session_id);
+    const list = this.inboundEvents.get(sessionKey(input.session_id)) ?? [];
+    return list
+      .filter(row => {
+        if (row.consumed) {
+          return false;
+        }
+        if (input.turn_id === undefined) {
+          return true;
+        }
+        if (input.turn_id === null) {
+          return row.turn_id === null;
+        }
+        return row.turn_id === input.turn_id;
+      })
+      .slice()
+      .sort((a, b) => (a.event_id < b.event_id ? -1 : a.event_id > b.event_id ? 1 : 0))
+      .map(row => ({
+        event_id: row.event_id,
+        turn_id: row.turn_id,
+        payload: deepCopy(row.payload),
+        created_at: row.created_at,
+      }));
+  }
+
+  async markSessionInboundEventsConsumed(input: MarkSessionInboundEventsConsumedInput): Promise<void> {
+    if (input.event_ids.length === 0) {
+      return;
+    }
+    this.requireSession(input.session_id);
+    const list = this.inboundEvents.get(sessionKey(input.session_id));
+    if (!list) {
+      return;
+    }
+    const wanted = new Set(input.event_ids);
+    for (const row of list) {
+      if (wanted.has(row.event_id)) {
+        row.consumed = true;
+      }
+    }
+  }
+
   /** Cost from turn metrics when present; duration is completed_at − created_at, floored at 0. */
   private addTerminalSessionMetrics(sessionId: string, created_at: Date, state: TerminalTurnState): void {
     const stored = this.sessions.get(sessionKey(sessionId));
@@ -499,6 +591,14 @@ export class InMemorySessionStore<
     stored.record.metrics.total_duration_ms += elapsed_ms > 0 ? Math.trunc(elapsed_ms) : 0;
   }
 
+  private requireSession(sessionId: string): StoredSession<TSessionCustom> {
+    const stored = this.sessions.get(sessionKey(sessionId));
+    if (!stored) {
+      throw new SessionNotFoundError(sessionId);
+    }
+    return stored;
+  }
+
   private requireTurn(sessionId: string, turnId: string): TurnRecord<TTurnCustom> {
     const turn = this.turns.get(turnKey({ session_id: sessionId, turn_id: turnId }));
     if (!turn) {

packages/trueforge-core/src/agent-session/store/SessionStoreErrors.ts

@@ -74,6 +74,18 @@ export class TurnAlreadyExistsError extends SessionStoreConflictError {
   }
 }
 
+export class SessionInboundEventAlreadyExistsError extends SessionStoreConflictError {
+  readonly session_id: string;
+  readonly event_id: string;
+
+  constructor(session_id: string, event_id: string, options?: ErrorOptions) {
+    super(`Session inbound event already exists: ${session_id}/${event_id}`, options);
+    this.name = 'SessionInboundEventAlreadyExistsError';
+    this.session_id = session_id;
+    this.event_id = event_id;
+  }
+}
+
 export class PreviousTurnRunningError extends SessionStoreConflictError {
   readonly previous_turn_id: string;