Abstract Class: DiscoveryOrchestrator<TRecord, TState>

Makaio Framework / ai-adapters-core/node / DiscoveryOrchestrator

Abstract Class: DiscoveryOrchestrator<TRecord, TState>

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:46

Abstract discovery orchestrator for shallow session discovery.

Overrides handleFileChange to call importer.extractDiscoveryMetadata(filePath) directly, bypassing the parseFile → validateRecords → extractSessionContext pipeline. Emits only adapter.session.discovered and saves a stub cursor (bytesRead: 0) so the full import starts from byte 0 when the user triggers lazy-load pick-up.

For files that have already been imported, the orchestrator delegates 'modified' events to the base class for incremental processing and manages the 'imported' ↔ 'tracking' status lifecycle.

See

BaseLogOrchestrator - Parent class with full import behavior

Extends

BaseLogOrchestrator<TRecord, TState>

Type Parameters

TRecord

TRecord

The adapter’s native log record type

TState

TState = unknown

The adapter’s resumable state type (default: unknown)

Constructors

Constructor

protected new DiscoveryOrchestrator<TRecord, TState>(config, importer): DiscoveryOrchestrator<TRecord, TState>

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:69

Parameters

config

LogOrchestratorConfig

importer

LogImporter<TRecord, TState>

Returns

DiscoveryOrchestrator<TRecord, TState>

Overrides

BaseLogOrchestrator.constructor

Properties

config

protected readonly config: Required<Omit<LogOrchestratorConfig, "directory" | "checkMakaioManaged">> & object

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:42

Type Declaration

checkMakaioManaged?

optional checkMakaioManaged?: (sessionId) => Promise<boolean>

Parameters

sessionId

string

Returns

Promise<boolean>

directory?

optional directory?: string

Inherited from

BaseLogOrchestrator.config

eventQueue

protected readonly eventQueue: LogImportEventQueue

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:48

Inherited from

BaseLogOrchestrator.eventQueue

importer

protected readonly importer: LogImporter<TRecord, TState>

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:54

Importer instance - subclasses create and manage their own typed instance.

Inherited from

BaseLogOrchestrator.importer

lastSeenMtimeMs

protected readonly lastSeenMtimeMs: Map<string, number>

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:64

Last observed mtime (ms) per tracked file path. Used by the polled handler to detect inactivity without re-reading disk.

logPrefix

abstract protected readonly logPrefix: string

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:51

Log prefix for console output - set by subclass

Inherited from

BaseLogOrchestrator.logPrefix

trackingFilePaths

protected readonly trackingFilePaths: Set<string>

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:51

Paths of files whose adapter sessions are currently in 'tracking' status. Maintained locally to avoid per-poll DB lookups.

trackingInactiveCount

protected readonly trackingInactiveCount: Map<string, number>

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:58

Per-file count of consecutive poll cycles with no mtime change while in 'tracking' status. Reset on each 'modified' event; incremented by the polled handler when mtime is unchanged.

watcher

protected readonly watcher: LogImportWatcher

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:47

Inherited from

BaseLogOrchestrator.watcher

Methods

buildCursorSessionContext()

protected buildCursorSessionContext(context): object & object

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:147

Builds the serialized session context for cursor persistence.

Parameters

context

LogImportSessionContext<TState>

The import session context to serialize.

Returns

Serialized cursor session context.

Inherited from

BaseLogOrchestrator.buildCursorSessionContext

clearTrackingState()

protected clearTrackingState(filePath): void

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:77

Clear all in-memory tracking collections for a single file path.

Parameters

filePath

string

File path to remove from local tracking state

Returns

void

dispose()

dispose(): Promise<void>

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:136

Returns

Promise<void>

Overrides

BaseLogOrchestrator.dispose

getLogFilePattern()

abstract protected getLogFilePattern(): string

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:98

Get the glob pattern for log files to watch.

Returns

string

Inherited from

BaseLogOrchestrator.getLogFilePattern

getMaxRecords()

protected getMaxRecords(): number | undefined

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:138

Returns the maximum number of records to parse per file. Override in subclasses to limit parsing (e.g., discovery mode).

Returns

number | undefined

Maximum record count, or undefined for no limit.

Inherited from

BaseLogOrchestrator.getMaxRecords

handleFileChange()

protected handleFileChange(event): Promise<void>

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:161

Simplified file change handler for discovery mode.

Dispatches to two paths:

No cursor exists → lightweight discovery (emit adapter.session.discovered, write stub cursor).
Cursor exists + changeType === 'modified' + session status is 'imported' or 'tracking' → delegate to base class for incremental import, then transition status to 'tracking' if it was 'imported'.

All other cases (e.g., cursor exists but status is 'discovered', or changeType is not 'modified') are silently skipped.

Parameters

event

LogFileChangeEvent

File change event from the watcher

Returns

Promise<void>

Overrides

BaseLogOrchestrator.handleFileChange

handleFirstRead()

protected handleFirstRead(filePath, records, bytesRead, mtime, isJsonFormat, startOffset, emitLifecycleEvents?): Promise<void>

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:351

Handle the first read of a log file.

Extracts session context, emits session lifecycle events, processes records, and saves the cursor. Subclasses may override to change behavior (e.g., DiscoveryOrchestrator skips message processing).

Parameters

filePath

string

Path to the log file

records

TRecord[]

Records parsed from the file

bytesRead

number

Bytes read during this parse

mtime

Date

File modification time

isJsonFormat

boolean

Whether the file uses JSON (mtime-based) format

startOffset

number

Byte offset this read started from

emitLifecycleEvents?

boolean = true

Whether to emit session and started lifecycle events for this pass

Returns

Promise<void>

Inherited from

BaseLogOrchestrator.handleFirstRead

handleIncrementalRead()

protected handleIncrementalRead(filePath, records, cursorContext, bytesRead, mtime, isJsonFormat, startOffset): Promise<void>

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:418

Handle an incremental read of a log file.

Processes new records since the last cursor position. Subclasses may override to skip incremental processing (e.g., DiscoveryOrchestrator).

Parameters

filePath

string

Path to the log file

records

TRecord[]

New records since last read

cursorContext

object & object

Existing cursor session context

bytesRead

number

Bytes read during this parse

mtime

Date

File modification time

isJsonFormat

boolean

Whether the file uses JSON (mtime-based) format

startOffset

number

Byte offset this read started from

Returns

Promise<void>

Inherited from

BaseLogOrchestrator.handleIncrementalRead

isEnabled()

isEnabled(): boolean

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:173

Returns

boolean

Inherited from

BaseLogOrchestrator.isEnabled

isRunning()

isRunning(): boolean

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:177

Returns

boolean

Inherited from

BaseLogOrchestrator.isRunning

isSessionSkipped()

protected isSessionSkipped(adapterSessionId): Promise<boolean>

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:514

Determine whether an adapter session should be skipped from import.

Delegates to the managed-session cache so concurrent checks for the same adapterSessionId share one storage lookup and tracked skips update stats.

Parameters

adapterSessionId

string

External adapter session ID being evaluated

Returns

Promise<boolean>

Promise resolving to true when the session is Makaio-managed and should be skipped

Inherited from

BaseLogOrchestrator.isSessionSkipped

maybeUpdateCursor()

protected maybeUpdateCursor(filePath, bytesRead, startOffset, mtime, isJsonFormat, sessionContext?): Promise<void>

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:493

Parameters

filePath

string

bytesRead

number

startOffset

number

mtime

Date

isJsonFormat

boolean

sessionContext?

object & object

Returns

Promise<void>

Inherited from

BaseLogOrchestrator.maybeUpdateCursor

onPollCycle()

protected onPollCycle(trackedFilePaths): Promise<void>

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:338

React to a completed poll cycle for inactivity-based tracking → imported transitions.

For each file in trackingFilePaths, checks if the watcher’s last-known mtime matches our stored lastSeenMtimeMs. If unchanged, increments the inactive counter. When the counter reaches the tracking inactivity threshold, the adapter session reverts to 'imported' and the file is removed from tracking.

Parameters

trackedFilePaths

ReadonlySet<string>

All file paths observed by the watcher in this cycle

Returns

Promise<void>

parseFile()

abstract protected parseFile(filePath, startOffset, maxRecords?): Promise<ParseFileResult<TRecord>>

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:118

Parse a log file from the given byte offset.

Parameters

filePath

string

Path to the log file

startOffset

number

Byte offset to start reading from

maxRecords?

number

Optional maximum number of records to return (for shallow discovery)

Returns

Promise<ParseFileResult<TRecord>>

Inherited from

BaseLogOrchestrator.parseFile

persistImportedStatus()

protected persistImportedStatus(filePath): Promise<boolean>

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:88

Persist the steady-state imported status for a tracked file.

Parameters

filePath

string

Absolute path to the tracked log file

Returns

Promise<boolean>

true when the file can be removed from local tracking state

queueCursorUpdate()

protected queueCursorUpdate(filePath, bytesRead, startOffset, mtime, isJsonFormat, sessionContext?, precedingEventPromises?): Promise<void>

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:549

Enqueue a cursor write as a PQueue task so it executes after all previously queued events for the same batch have been emitted.

PQueue runs tasks with concurrency=1 in FIFO order, so placing the cursor write at the back of the queue after all queueEvent calls guarantees that the cursor advances only once every preceding event in the batch has been delivered. This prevents the race where a process exit after queueEvent but before queue drain would leave events lost yet the cursor advanced.

Parameters

filePath

string

Path to the log file

bytesRead

number

Total bytes read at the new cursor position

startOffset

number

Byte offset this read started from

mtime

Date

File modification time

isJsonFormat

boolean

Whether the file uses JSON (mtime-based) format

sessionContext?

object & object

Serialized session context for the cursor

precedingEventPromises?

Promise<void>[] = []

Emission promises queued before this cursor write

Returns

Promise<void>

Promise that resolves when the cursor write has completed

Inherited from

BaseLogOrchestrator.queueCursorUpdate

queueEvent()

protected queueEvent(event): Promise<void>

Defined in: ../../../adapters/core/src/log-importer/base-orchestrator.ts:527

Queue a normalized event and return its delivery promise.

Parameters

event

NormalizedEvent

Normalized event to emit

Returns

Promise<void>

Promise that resolves after delivery or rejects on emit failure

Inherited from

BaseLogOrchestrator.queueEvent

restorePersistedTrackingState()

protected restorePersistedTrackingState(): Promise<void>

Defined in: ../../../adapters/core/src/log-importer/discovery-orchestrator.ts:193

Rehydrate tracking file state from persisted adapter-session rows after startup.

The watcher state is process-local, but adapter-session status is persisted. Without this rehydration, a restart can strand sessions in 'tracking' forever because no in-memory file state exists to drive the inactivity timeout.