packages/cli/src/acp/session-event-mapper.ts

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

import { foldRuntimeHostAssistantDelta } from '@maka/runtime-host/adapter';
import {
  RequestError,
  type SessionNotification,
  type SessionUpdate,
} from '@agentclientprotocol/sdk';
import type { SessionEvent } from '@maka/core/events';
import type { StoredMessage } from '@maka/core/session';
import type { InteractionPendingSnapshot, InteractionSnapshot } from '@maka/runtime-host/protocol';
import { whileActive } from './active-promise.js';
import { AcpToolEventMapper } from './tool-event-mapper.js';

type StreamKind = 'text' | 'thinking';

export interface AcpSessionEventMapperOptions {
  readonly sessionId: string;
  readonly notify: (notification: SessionNotification) => Promise<void>;
  /** Ends projection delivery without waiting for a stalled client transport. */
  readonly signal?: AbortSignal;
}

/** Serializes one ACP prompt's live projection delivery. */
export class AcpSessionEventMapper {
  readonly #sessionId: string;
  readonly #notify: (notification: SessionNotification) => Promise<void>;
  readonly #signal: AbortSignal | undefined;
  readonly #streams = new Map<string, string>();
  readonly #tools: AcpToolEventMapper;
  #tail: Promise<unknown> = Promise.resolve();
  #failure: unknown;
  #failed = false;

  constructor(options: AcpSessionEventMapperOptions) {
    this.#sessionId = options.sessionId;
    this.#notify = options.notify;
    this.#signal = options.signal;
    this.#tools = new AcpToolEventMapper((update) =>
      this.#deliver({ sessionId: this.#sessionId, update }),
    );
  }

  accept(event: SessionEvent): Promise<void> {
    return this.#enqueue(async () => {
      switch (event.type) {
        case 'text_delta':
          await this.#acceptText(
            'text',
            event.messageId,
            deltaText(event, this.#streams.get(streamKey('text', event.messageId))),
          );
          break;
        case 'text_complete':
          await this.#acceptText('text', event.messageId, event.text);
          break;
        case 'thinking_delta':
          await this.#acceptText(
            'thinking',
            event.messageId,
            deltaText(event, this.#streams.get(streamKey('thinking', event.messageId))),
          );
          break;
        case 'thinking_complete':
          await this.#acceptText('thinking', event.messageId, event.text);
          break;
        case 'tool_start':
        case 'tool_output_delta':
        case 'tool_progress':
        case 'tool_result_preview':
        case 'tool_result':
          await this.#tools.accept(event);
          break;
        default:
          break;
      }
    });
  }

  replaceTranscript(turnId: string, messages: readonly StoredMessage[]): Promise<void> {
    return this.acceptTranscriptMessages(turnId, messages);
  }

  /** Apply a bounded authoritative batch; absence from one batch never removes a tool. */
  acceptTranscriptMessages(turnId: string, messages: readonly StoredMessage[]): Promise<void> {
    return this.#enqueue(async () => {
      for (const message of messages) {
        if (message.turnId !== turnId) continue;
        if (message.type === 'assistant') {
          await this.#acceptText('thinking', message.id, message.thinking?.text ?? '');
          await this.#acceptText('text', message.id, message.text);
        } else await this.#tools.acceptMessage(message);
      }
    });
  }

  finishTools(
    turnId: string,
    terminalStatus: 'completed' | 'failed' | 'cancelled' = 'completed',
  ): Promise<void> {
    return this.#enqueue(() => this.#tools.finishTools(turnId, terminalStatus));
  }

  pendingInteraction(pending: InteractionPendingSnapshot): Promise<void> {
    return this.#enqueue(() => this.#tools.pendingInteraction(pending));
  }

  resolvedInteraction(
    resolved: InteractionSnapshot,
    pending: InteractionPendingSnapshot,
  ): Promise<void> {
    return this.#enqueue(() => this.#tools.resolvedInteraction(resolved, pending));
  }

  /** Waits until every notification already accepted by this mapper has settled. */
  flush(): Promise<void> {
    return this.#tail.then(() => {
      if (this.#failed) throw this.#failure;
    });
  }

  async #acceptText(kind: StreamKind, hostMessageId: string, nextText: string): Promise<void> {
    const key = streamKey(kind, hostMessageId);
    const current = this.#streams.get(key) ?? '';
    if (!nextText.startsWith(current)) {
      // ACP v1 chunks only append. A new message ID cannot retract prior output.
      this.#failure = RequestError.internalError(
        { source: 'adapter', code: 'unsupported_stream_revision' },
        'Runtime Host revised streamed output that ACP v1 cannot replace; the prompt failed',
      );
      throw this.#failure;
    }
    const chunk = nextText.slice(current.length);
    this.#streams.set(key, nextText);
    if (chunk.length === 0) return;
    const update: SessionUpdate = {
      sessionUpdate: kind === 'text' ? 'agent_message_chunk' : 'agent_thought_chunk',
      content: { type: 'text', text: chunk },
      messageId: hostMessageId,
    };
    await this.#deliver({ sessionId: this.#sessionId, update });
  }

  async #deliver(notification: SessionNotification): Promise<void> {
    if (this.#signal?.aborted) return;
    const delivery = this.#notify(notification);
    if (!this.#signal) return delivery;
    await whileActive(delivery, this.#signal);
  }

  #enqueue<T>(operation: () => Promise<T>): Promise<T> {
    const result = this.#tail.then(async () => {
      if (this.#failed) throw this.#failure;
      try {
        return await operation();
      } catch (error) {
        this.#failure = error;
        this.#failed = true;
        throw error;
      }
    });
    this.#tail = result.then(
      () => undefined,
      () => undefined,
    );
    return result;
  }
}

function deltaText(
  event: Extract<SessionEvent, { type: 'text_delta' | 'thinking_delta' }>,
  current = '',
): string {
  return foldRuntimeHostAssistantDelta(current, {
    startOffset: event.startOffset ?? current.length,
    text: event.text,
  }).text;
}

function streamKey(kind: StreamKind, messageId: string): string {
  return `${kind}:${messageId}`;
}