fix(lark): ingest inbound images and videos as chat attachments (MUL-4934)

server/cmd/server/main.go

@@ -498,7 +498,11 @@ func main() {
 			)
 		}
 		if h.ChannelRouter != nil {
-			h.ChannelRouter.Drain()
+			drainCtx, drainCancel := context.WithTimeout(context.Background(), 10*time.Second)
+			if !h.ChannelRouter.Drain(drainCtx) {
+				slog.Warn("channel router: drain deadline reached; deferred media fallback remains durable")
+			}
+			drainCancel()
 		}
 	}
 

server/cmd/server/router.go

@@ -391,8 +391,9 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus
 					Credentials: installSvc,
 					Logger:      slog.Default(),
 				})
+				mediaResolver := lark.NewFeishuMediaResolver(larkClient, installSvc, store, slog.Default())
 				channelRouter.Register(channel.TypeFeishu, lark.NewFeishuResolverSet(
-					cs, feishuSession, auditLogger, resolverReplier, typingIndicator,
+					cs, feishuSession, auditLogger, resolverReplier, typingIndicator, mediaResolver,
 				))
 				slog.Info("lark inbound pipeline wired", "connector", connectorLabel)
 

server/internal/handler/daemon.go

@@ -2644,6 +2644,9 @@ func (h *Handler) ResolveTaskSkillBundles(w http.ResponseWriter, r *http.Request
 // these, not just the latest. Every completed or failed run writes an
 // assistant row, so the anchor advances one turn at a time; the result is the
 // whole slice on the first turn and exactly the new message(s) thereafter.
+// Legacy channel tasks also stop at the first unexpired media marker so an old
+// run cannot consume a placeholder that has not been bound yet. New channel
+// tasks use task-owned input batches and do not depend on this fallback.
 func trailingUserMessages(msgs []db.ChatMessage) []db.ChatMessage {
 	start := 0
 	for i := len(msgs) - 1; i >= 0; i-- {
@@ -2652,7 +2655,15 @@ func trailingUserMessages(msgs []db.ChatMessage) []db.ChatMessage {
 			break
 		}
 	}
-	return msgs[start:]
+	msgs = msgs[start:]
+	now := time.Now()
+	for i := range msgs {
+		pending := msgs[i].ChannelMediaPendingUntil
+		if pending.Valid && pending.Time.After(now) {
+			return msgs[:i]
+		}
+	}
+	return msgs
 }
 
 // ListPendingTasksByRuntime returns queued/dispatched tasks for a runtime.

server/internal/handler/daemon_chat_prompt_test.go

@@ -2,14 +2,24 @@ package handler
 
 import (
 	"testing"
+	"time"
 
+	"github.com/jackc/pgx/v5/pgtype"
 	db "github.com/multica-ai/multica/server/pkg/db/generated"
 )
 
 func msg(role, content string) db.ChatMessage {
 	return db.ChatMessage{Role: role, Content: content}
 }
 
+func mediaPendingMsg(content string, until time.Time) db.ChatMessage {
+	return db.ChatMessage{
+		Role:                     "user",
+		Content:                  content,
+		ChannelMediaPendingUntil: pgtype.Timestamptz{Time: until, Valid: true},
+	}
+}
+
 func contents(msgs []db.ChatMessage) []string {
 	out := make([]string, len(msgs))
 	for i, m := range msgs {
@@ -76,6 +86,23 @@ func TestTrailingUserMessages(t *testing.T) {
 			in:   []db.ChatMessage{msg("user", "hi")},
 			want: []string{"hi"},
 		},
+		{
+			name: "stops before media that is still pending",
+			in: []db.ChatMessage{
+				msg("user", "ready"),
+				mediaPendingMsg("[Image]", time.Now().Add(time.Minute)),
+				msg("user", "after image"),
+			},
+			want: []string{"ready"},
+		},
+		{
+			name: "expired media placeholder is ready",
+			in: []db.ChatMessage{
+				mediaPendingMsg("[Image]", time.Now().Add(-time.Minute)),
+				msg("user", "after image"),
+			},
+			want: []string{"[Image]", "after image"},
+		},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {

server/internal/integrations/channel/engine/resolvers.go

@@ -89,23 +89,40 @@ type EnsureSessionParams struct {
 // AppendParams carries the inputs for SessionBinder.AppendMessage. ClaimToken
 // is the dedup owner-fence token; the binder runs the dedup Mark INSIDE its
 // chat_message+session tx so the durable write and the Mark commit atomically.
+// MediaPendingUntil persists the placeholder fallback deadline.
 type AppendParams struct {
-	SessionID      pgtype.UUID
-	Sender         pgtype.UUID
-	InstallationID pgtype.UUID
-	Message        channel.InboundMessage
-	ClaimToken     pgtype.UUID
+	SessionID         pgtype.UUID
+	Sender            pgtype.UUID
+	InstallationID    pgtype.UUID
+	Message           channel.InboundMessage
+	ClaimToken        pgtype.UUID
+	MediaPendingUntil pgtype.Timestamptz
 }
 
 // AppendResult reports what AppendMessage decided.
 type AppendResult struct {
+	// MessageID is the durable chat_message row created by AppendMessage.
+	// Detached media processing uses it to link attachments after the
+	// connector ACK path has completed.
+	MessageID pgtype.UUID
 	// IssueCommand is non-nil when the message was an /issue command.
 	IssueCommand *IssueCommand
 	// DedupMarked is true when AppendMessage finalized the dedup claim in its
 	// own tx; the Router then skips the post-pipeline finalize.
 	DedupMarked bool
 }
 
+// BindMediaParams carries stored media references to the post-append
+// attachment transaction. MessageID is the durable chat_message created by
+// AppendMessage; media downloads must never run inside this transaction.
+type BindMediaParams struct {
+	MessageID   pgtype.UUID
+	SessionID   pgtype.UUID
+	WorkspaceID pgtype.UUID
+	Sender      pgtype.UUID
+	MediaRefs   []channel.MediaRef
+}
+
 // IssueCommand is the parsed /issue command.
 type IssueCommand struct {
 	Title       string
@@ -161,6 +178,23 @@ type Deduper interface {
 type SessionBinder interface {
 	EnsureSession(ctx context.Context, p EnsureSessionParams) (pgtype.UUID, error)
 	AppendMessage(ctx context.Context, p AppendParams) (AppendResult, error)
+	BindMedia(ctx context.Context, p BindMediaParams) error
+}
+
+// MediaResolver resolves platform media after the user message and dedup mark
+// are durable. The Router runs it off the connector ACK path and binds any
+// returned MediaRefs; the independently scheduled task remains deferred until
+// binding finishes or the persisted deadline expires. Implementations are
+// best-effort: failures leave the stored placeholder text intact.
+type MediaResolver interface {
+	// HasMedia reports whether msg references platform media that
+	// ResolveMedia would fetch. The Router calls it synchronously on the
+	// connector ACK path to decide whether to persist a media deadline and
+	// queue a resolution job at all, so implementations must be pure
+	// in-memory checks (no I/O). A false result keeps the message on the
+	// plain ingest path: no marker, no deferred run, no semaphore slot.
+	HasMedia(msg channel.InboundMessage) bool
+	ResolveMedia(ctx context.Context, inst ResolvedInstallation, sender ResolvedIdentity, sessionID pgtype.UUID, msg channel.InboundMessage) channel.InboundMessage
 }
 
 // Auditor records a dropped inbound event (no message body — drop-audit
@@ -200,6 +234,7 @@ type ResolverSet struct {
 	Identity     IdentityResolver
 	Dedup        Deduper
 	Session      SessionBinder
+	Media        MediaResolver
 	Audit        Auditor
 	Replier      OutboundReplier
 	Typing       TypingNotifier
@@ -216,6 +251,7 @@ type IssueCreator interface {
 // trigger a chat run. Shared across platforms.
 type TaskEnqueuer interface {
 	EnqueueChatTask(ctx context.Context, session db.ChatSession, initiatorUserID pgtype.UUID, forceFreshSession bool) (db.AgentTaskQueue, error)
+	PromoteChannelChatTasksIfMediaReady(ctx context.Context, sessionID pgtype.UUID) error
 }
 
 // SessionReader reads the rows the debounced flush + /issue identifier need.