createSlackReplyDeliveryPlan,

deliverReplies,

readSlackReplyBlocks,

resolveDeliveredSlackReplyThreadTs,

resolveSlackThreadTs,

} from "../replies.js";

import { createReplyDispatcherWithTyping, dispatchInboundMessage } from "../reply.runtime.js";

let streamSession: SlackStreamSession | null = null;

let streamFailed = false;

let usedReplyThreadTs: string | undefined;

let usedBlockReplyThreadTs: string | undefined;

let observedReplyDelivery = false;

const deliveryTracker = createSlackTurnDeliveryTracker();

const resolveDeliveryThreadTs = (params: {

kind: ReplyDispatchKind;

forcedThreadTs?: string;

}): string | undefined => {

const plannedThreadTs = params.forcedThreadTs ? undefined : replyPlan.nextThreadTs();

return (

params.forcedThreadTs ??

plannedThreadTs ??

(params.kind === "block" ? usedBlockReplyThreadTs : undefined)

);

};

const rememberDeliveredThreadTs = (

kind: ReplyDispatchKind,

deliveredThreadTs: string | undefined,

) => {

if (!deliveredThreadTs) {

return;

}

usedReplyThreadTs ??= deliveredThreadTs;

if (kind === "block") {

usedBlockReplyThreadTs = deliveredThreadTs;

}

};

const deliverPendingStreamFallback = async (

session: SlackStreamSession,

err: SlackStreamNotDeliveredError,

kind: ReplyDispatchKind;

forcedThreadTs?: string;

}): Promise<void> => {

const replyThreadTs = params.forcedThreadTs ?? replyPlan.nextThreadTs();

const replyThreadTs = resolveDeliveryThreadTs(params);

if (

deliveryTracker.hasDelivered({

kind: params.kind,

...(slackIdentity ? { identity: slackIdentity } : {}),

});

observedReplyDelivery = true;

const deliveredThreadTs = resolveDeliveredSlackReplyThreadTs({

replyToMode: prepared.replyToMode,

payloadReplyToId: params.payload.replyToId,

replyThreadTs,

});

// Record the thread ts only after confirmed delivery success.

if (replyThreadTs) {

usedReplyThreadTs ??= replyThreadTs;

}

rememberDeliveredThreadTs(params.kind, deliveredThreadTs);

replyPlan.markSent();

deliveryTracker.markDelivered({

kind: params.kind,

threadTs: params.session.threadTs,

textOverride: params.textOverride,

});

rememberDeliveredThreadTs(params.kind, params.session.threadTs);

return true;

};

"slack-stream: no reply thread target for stream start, falling back to normal delivery",

);

streamFailed = true;

await deliverNormally({ payload: params.payload, kind: params.kind });

await deliverNormally({

payload: params.payload,

kind: params.kind,

});

return;

}

if (

if (streamSession.delivered) {

observedReplyDelivery = true;

}

usedReplyThreadTs ??= streamThreadTs;

rememberDeliveredThreadTs(params.kind, streamThreadTs);

replyPlan.markSent();

deliveryTracker.markDelivered({

kind: params.kind,

});

},

deliverNormally: async () => {

await deliverNormally({ payload, kind: info.kind });

await deliverNormally({

payload,

kind: info.kind,

});

},

onPreviewFinalized: (_preview) => {

const finalThreadTs = usedReplyThreadTs ?? statusThreadTs;