Abstract Class: BaseLogOrchestrator<TRecord, TState>

Makaio Framework

---

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

Abstract Class: BaseLogOrchestrator<TRecord, TState>

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

Abstract base class for log import orchestrators.

Supports JSONL (byte offset cursors) and JSON (mtime-based) file formats. Concrete implementations provide adapter-specific file parsing.

Extended by

• DiscoveryOrchestrator

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 BaseLogOrchestrator<TRecord, TState>(config, importer): BaseLogOrchestrator<TRecord, TState>

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

Parameters

config

LogOrchestratorConfig

importer

LogImporter<TRecord, TState>

Returns

BaseLogOrchestrator<TRecord, TState>

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

---

eventQueue

protected readonly eventQueue: LogImportEventQueue

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

---

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.

---

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

---

watcher

protected readonly watcher: LogImportWatcher

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

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.

---

dispose()

dispose(): Promise<void>

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

Returns

Promise<void>

---

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

---

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.

---

handleFileChange()

protected handleFileChange(event): Promise<void>

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

Handle a file-change event: resolve cursor state and route to first-read or incremental-read as appropriate.

Subclasses (e.g., DiscoveryOrchestrator) may override this method to substitute an entirely different dispatch strategy.

Parameters

event

LogFileChangeEvent

File change event from the watcher

Returns

Promise<void>

---

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>

---

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>

---

isEnabled()

isEnabled(): boolean

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

Returns

boolean

---

isRunning()

isRunning(): boolean

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

Returns

boolean

---

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

---

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>

---

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>>

---

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

---

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

---

shouldSkipFile()

protected shouldSkipFile(_filePath): boolean

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

Determine whether a discovered file should be skipped before processing.

Default implementation never skips. Subclasses may override to exclude files by name (e.g., ephemeral compaction summary files).

Parameters

_filePath

string

Absolute path to the candidate file

Returns

boolean

true to skip the file entirely, false to process normally

---

start()

start(): Promise<void>

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

Returns

Promise<void>

---

stop()

stop(): Promise<void>

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

Returns

Promise<void>

---

trackFileChange()

protected trackFileChange(event): void

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

Track a watcher-triggered file import so shutdown can wait for any cursor work that the handler enqueues before draining the shared FIFO queue.

Parameters

event

LogFileChangeEvent

File change event to process

Returns

void

---

trackImportedSession()

protected trackImportedSession(adapterSessionId): void

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

Record that a session has been imported, updating import statistics.

Protected to allow subclasses to update stats when processing additional sessions (e.g., compress child sessions from compacted files).

Parameters

adapterSessionId

string

The session ID that was imported

Returns

void

---

updateCursor()

protected updateCursor(filePath, bytesRead, mtime, sessionContext?): Promise<void>

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

Parameters

filePath

string

bytesRead

number

mtime

Date

sessionContext?

object & object

Returns

Promise<void>

---

usesJsonFormat()

protected usesJsonFormat(): boolean

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

Check if this orchestrator uses JSON format (mtime-based cursor).

Returns

boolean

True if JSON format, false for JSONL

---

validateRecords()

protected validateRecords(records): TRecord[]

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

Validate and filter parsed records. Default: returns records as-is.

Parameters

records

TRecord[]

Raw parsed records

Returns

TRecord[]

Validated/filtered records

---

createDefaultCheckMakaioManaged()

static createDefaultCheckMakaioManaged(): (sessionId) => Promise<boolean>

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

Returns

Function that checks if a session is Makaio-managed.

(sessionId) => Promise<boolean>