Method fetchRecentLogs

backend/src/trace/log-stream.service.ts:121–181

(
    runId: string,
    organizationId?: string | null,
    lastCursor?: string | null,
  )

  }

  async fetchRecentLogs(
    runId: string,
    organizationId?: string | null,
    lastCursor?: string | null,
  ): Promise<{
    logs: {
      id: string;
      runId: string;
      nodeId: string;
      level: string;
      message: string;
      timestamp: string;
      sequence: number;
    }[];
    cursor: string | null;
  }> {
    const selector = this.buildSelector({ run_id: runId });
    const startTime =
      lastCursor ??
      (await this.resolveEarliestTimestamp(runId, organizationId))?.toISOString() ??
      new Date(Date.now() - this.defaultTailWindowMs).toISOString();
    const endTime = new Date().toISOString();

    const entries = await this.queryLokiTimeRange(selector, startTime, endTime, 500);
    const filtered = entries.filter((entry) => {
      if (!lastCursor) {
        return true;
      }
      const ts = Date.parse(entry.timestamp);
      const lastTs = Date.parse(lastCursor);
      if (Number.isNaN(ts) || Number.isNaN(lastTs)) {
        return true;
      }
      return ts > lastTs;
    });

    if (filtered.length === 0) {
      return { logs: [], cursor: lastCursor ?? startTime ?? null };
    }

    const logs = filtered.map((entry, index) => {
      const timestampValue = Date.parse(entry.timestamp);
      const sequence =
        Number.isNaN(timestampValue) || timestampValue < 0 ? Date.now() + index : timestampValue;
      return {
        id: `${runId}-${sequence}-${index}`,
        runId,
        nodeId: entry.nodeId || 'unknown',
        level: entry.level || 'info',
        message: entry.message,
        timestamp: entry.timestamp,
        sequence,
      };
    });

    const nextCursor = logs[logs.length - 1]?.timestamp ?? lastCursor ?? startTime ?? null;
    return {
      logs,