feat(lark): support inbound message attachments

server/cmd/server/main.go

@@ -428,6 +428,9 @@ func main() {
 	if h.WebhookDeliveryWorker != nil {
 		go h.WebhookDeliveryWorker.Run(sweepCtx)
 	}
+	if h.LarkInboundDeliveryWorker != nil {
+		go h.LarkInboundDeliveryWorker.Run(sweepCtx)
+	}
 
 	// Channel inbound supervisor (MUL-3620): holds the §4.4 WS lease per
 	// installation and drives each channel.Channel. It is built
@@ -537,6 +540,9 @@ func main() {
 				"timeout", h.ChannelSupervisor.ShutdownTimeout().String(),
 			)
 		}
+		if h.LarkInboundDeliveryWorker != nil && !h.LarkInboundDeliveryWorker.WaitWithTimeout(5*time.Second) {
+			slog.Warn("lark inbound delivery worker did not exit within shutdown timeout")
+		}
 		if h.ChannelRouter != nil {
 			h.ChannelRouter.Drain()
 		}

server/cmd/server/router.go

@@ -384,15 +384,26 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus
 				// Registering the Factory (connect/send) + ResolverSet
 				// (inbound pipeline seams) is all it takes to add the platform
 				// to the engine — no engine edit.
-				connector, connectorLabel := buildLarkConnector(installSvc, larkClient)
+				enricher := lark.NewInboundEnricher(larkClient, lark.InboundEnricherConfig{
+					RecentContextSize: lark.DefaultRecentContextSize,
+					Logger:            slog.Default(),
+				})
+				inboundWorker := lark.NewInboundDeliveryWorker(queries, installSvc, enricher, channelRouter.Handle, slog.Default())
+				h.LarkInboundDeliveryWorker = inboundWorker
+				connector, connectorLabel := buildLarkConnector(installSvc)
 				lark.RegisterFeishu(channelRegistry, lark.FeishuChannelDeps{
 					Connector:   connector,
+					Inbound:     inboundWorker,
 					APIClient:   larkClient,
 					Credentials: installSvc,
 					Logger:      slog.Default(),
 				})
+				var mediaResolver *lark.FeishuMediaResolver
+				if downloader, ok := larkClient.(lark.MessageResourceDownloader); ok && store != nil {
+					mediaResolver = lark.NewFeishuMediaResolver(downloader, installSvc, store, slog.Default())
+				}
 				channelRouter.Register(channel.TypeFeishu, lark.NewFeishuResolverSet(
-					cs, feishuSession, auditLogger, resolverReplier, typingIndicator,
+					cs, feishuSession, auditLogger, mediaResolver, resolverReplier, typingIndicator,
 				))
 				slog.Info("lark inbound pipeline wired", "connector", connectorLabel)
 
@@ -1412,7 +1423,7 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus
 //
 // Returns the connector plus a short label for the boot log:
 // "ws-long-conn" in the healthy case, "noop" in the fallback case.
-func buildLarkConnector(installSvc *lark.InstallationService, apiClient lark.APIClient) (lark.EventConnector, string) {
+func buildLarkConnector(installSvc *lark.InstallationService) (lark.EventConnector, string) {
 	endpointFetcher, err := lark.NewHTTPConnectionTokenFetcher(lark.HTTPConnectionTokenConfig{
 		BaseURL: strings.TrimSpace(os.Getenv("MULTICA_LARK_CALLBACK_BASE_URL")),
 		Logger:  slog.Default(),
@@ -1441,20 +1452,10 @@ func buildLarkConnector(installSvc *lark.InstallationService, apiClient lark.API
 		}
 		return creds, nil
 	})
-	// Inbound enricher: expands quoted replies / forwarded bundles AND
-	// prefetches a window of surrounding group history (MUL-3084) into the
-	// agent's body via the IM API before dispatch. It shares the
-	// connector's resolved credentials and runs under the connector's
-	// EnrichTimeout so it cannot overrun the Lark long-conn ACK budget.
-	enricher := lark.NewInboundEnricher(apiClient, lark.InboundEnricherConfig{
-		RecentContextSize: lark.DefaultRecentContextSize,
-		Logger:            slog.Default(),
-	})
 	conn, err := lark.NewWSLongConnConnector(lark.WSConnectorConfig{
 		Dialer:              dialer,
 		EndpointFetcher:     endpointFetcher,
 		FrameDecoder:        decoder,
-		Enricher:            enricher,
 		CredentialsProvider: credsProvider,
 		Logger:              slog.Default(),
 	})

server/internal/handler/file_test.go

@@ -69,6 +69,14 @@ func (m *mockStorage) Upload(_ context.Context, key string, data []byte, _ strin
 	return fmt.Sprintf("https://cdn.example.com/%s", key), nil
 }
 
+func (m *mockStorage) UploadFromReader(ctx context.Context, key string, reader io.Reader, _ int64, contentType string, filename string) (string, error) {
+	data, err := io.ReadAll(reader)
+	if err != nil {
+		return "", err
+	}
+	return m.Upload(ctx, key, data, contentType, filename)
+}
+
 func (m *mockStorage) Delete(_ context.Context, key string) {
 	m.mu.Lock()
 	defer m.mu.Unlock()

server/internal/handler/handler.go

@@ -185,6 +185,10 @@ type Handler struct {
 	// UI consults IsConfigured() to decide whether to surface install
 	// entry points.
 	LarkAPIClient lark.APIClient
+	// LarkInboundDeliveryWorker owns the durable receive queue. Feishu ACKs
+	// after Enqueue commits; enrichment, media download, and Router dispatch
+	// happen here under a recoverable Postgres lease.
+	LarkInboundDeliveryWorker *lark.InboundDeliveryWorker
 	// Composio integration (MUL-3720). Nil when COMPOSIO_API_KEY is unset;
 	// the composio HTTP handlers return 503 in that case. Wired in
 	// cmd/server/router.go after handler.New.

server/internal/integrations/channel/channel.go

@@ -3,6 +3,8 @@ package channel
 import (
 	"context"
 	"encoding/json"
+
+	"github.com/jackc/pgx/v5/pgtype"
 )
 
 // Type identifies an inbound channel platform — the discriminator the
@@ -81,8 +83,9 @@ type Channel interface {
 // the channel_type column + JSONB config of a channel_installation row
 // (MUL-3515 decision §3).
 type Config struct {
-	Type Type
-	Raw  json.RawMessage
+	InstallationID pgtype.UUID
+	Type           Type
+	Raw            json.RawMessage
 
 	// Handler is the shared inbound entry point the engine injects so the
 	// built Channel can deliver normalized InboundMessage values into the

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

@@ -90,6 +90,7 @@ type EnsureSessionParams struct {
 // 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.
 type AppendParams struct {
+	WorkspaceID    pgtype.UUID
 	SessionID      pgtype.UUID
 	Sender         pgtype.UUID
 	InstallationID pgtype.UUID
@@ -146,6 +147,14 @@ type IdentityResolver interface {
 	ResolveSender(ctx context.Context, inst ResolvedInstallation, msg channel.InboundMessage) (ResolvedIdentity, error)
 }
 
+// MediaResolver persists platform-bound message resources after every access
+// gate has passed and before the user message transaction starts. Cleanup is
+// called when a later persistence step fails; a successful append owns the
+// objects and the callback is discarded.
+type MediaResolver interface {
+	Resolve(ctx context.Context, inst ResolvedInstallation, identity ResolvedIdentity, msg channel.InboundMessage) (resolved channel.InboundMessage, cleanup func(context.Context), err error)
+}
+
 // Deduper is the two-phase idempotency seam. Claim mints an owner-fence token
 // (ErrDuplicate when already processed / in flight); Mark/Release are fenced on
 // the token (a no-op on token mismatch is not an error).
@@ -198,6 +207,7 @@ type TypingNotifier interface {
 type ResolverSet struct {
 	Installation InstallationResolver
 	Identity     IdentityResolver
+	Media        MediaResolver
 	Dedup        Deduper
 	Session      SessionBinder
 	Audit        Auditor

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

@@ -242,7 +242,27 @@ func (r *Router) processClaimed(ctx context.Context, set ResolverSet, msg channe
 		}
 	}
 
-	// 5. Resolve the chat_session. Group sessions are created by the INSTALLER
+	// 5. Resolve platform media only after installation, group-addressing,
+	// identity, and live membership checks have passed. This prevents an
+	// untrusted/unbound sender from turning a resource key into network I/O.
+	var mediaCleanup func(context.Context)
+	if set.Media != nil {
+		resolved, cleanup, resolveErr := set.Media.Resolve(ctx, inst, identity, msg)
+		if resolveErr != nil {
+			runMediaCleanup(cleanup)
+			return Result{}, finalizeRelease, fmt.Errorf("resolve media: %w", resolveErr)
+		}
+		msg = resolved
+		mediaCleanup = cleanup
+	}
+	mediaCommitted := false
+	defer func() {
+		if !mediaCommitted {
+			runMediaCleanup(mediaCleanup)
+		}
+	}()
+
+	// 6. Resolve the chat_session. Group sessions are created by the INSTALLER
 	//    (stable workspace identity that won't churn with group membership);
 	//    p2p sessions by the sole human sender.
 	sessionCreator := identity.UserID
@@ -259,8 +279,9 @@ func (r *Router) processClaimed(ctx context.Context, set ResolverSet, msg channe
 		return Result{}, finalizeRelease, fmt.Errorf("ensure chat session: %w", err)
 	}
 
-	// 6. Append message + in-tx dedup Mark — the durable transition point.
+	// 7. Append message + in-tx dedup Mark — the durable transition point.
 	appendRes, err := set.Session.AppendMessage(ctx, AppendParams{
+		WorkspaceID:    inst.WorkspaceID,
 		SessionID:      sessionID,
 		Sender:         identity.UserID,
 		InstallationID: inst.ID,
@@ -273,6 +294,7 @@ func (r *Router) processClaimed(ctx context.Context, set ResolverSet, msg channe
 		}
 		return Result{}, finalizeRelease, fmt.Errorf("append user message: %w", err)
 	}
+	mediaCommitted = true
 
 	// Post-append paths must NOT Release (chat_message + Mark already
 	// committed). Mark-again is a no-op, so finalizeNone — unless the binder
@@ -289,7 +311,7 @@ func (r *Router) processClaimed(ctx context.Context, set ResolverSet, msg channe
 		Sender:         msg.Source.SenderID,
 	}
 
-	// 7. /issue command, if present. chat_message is already durable; all
+	// 8. /issue command, if present. chat_message is already durable; all
 	//    error returns from here signal finalizeNone (or the defensive Mark).
 	if appendRes.IssueCommand != nil {
 		issueRes, err := r.createIssue(ctx, inst, set.OriginType, identity.UserID, sessionID, *appendRes.IssueCommand)
@@ -306,7 +328,7 @@ func (r *Router) processClaimed(ctx context.Context, set ResolverSet, msg channe
 		}
 	}
 
-	// 8. Debounce the run trigger. The synchronous outcome is OutcomeIngested
+	// 9. Debounce the run trigger. The synchronous outcome is OutcomeIngested
 	//    with no TaskID — the task row is created at flush. identity.UserID is
 	//    THIS message's sender (the task initiator), deliberately not the
 	//    session creator (group sessions are creator=installer). Latest sender
@@ -315,6 +337,15 @@ func (r *Router) processClaimed(ctx context.Context, set ResolverSet, msg channe
 	return res, postAppendFinalize, nil
 }
 
+func runMediaCleanup(cleanup func(context.Context)) {
+	if cleanup == nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	cleanup(ctx)
+}
+
 // scheduleRun hands the per-session run trigger to the debouncer (or fires it
 // inline when batching is disabled).
 func (r *Router) scheduleRun(set ResolverSet, inst ResolvedInstallation, msg channel.InboundMessage, sessionID, initiatorUserID pgtype.UUID) {

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

@@ -30,6 +30,21 @@ type fakeIdentity struct {
 	err error
 }
 
+type fakeMediaResolver struct {
+	calls        int
+	cleanupCalls int
+	result       channel.InboundMessage
+	err          error
+}
+
+func (f *fakeMediaResolver) Resolve(_ context.Context, _ ResolvedInstallation, _ ResolvedIdentity, msg channel.InboundMessage) (channel.InboundMessage, func(context.Context), error) {
+	f.calls++
+	if f.result.MessageID == "" {
+		f.result = msg
+	}
+	return f.result, func(context.Context) { f.cleanupCalls++ }, f.err
+}
+
 func (f *fakeIdentity) ResolveSender(_ context.Context, _ ResolvedInstallation, _ channel.InboundMessage) (ResolvedIdentity, error) {
 	return f.id, f.err
 }
@@ -216,6 +231,7 @@ type harness struct {
 	router  *Router
 	inst    *fakeInstaller
 	ident   *fakeIdentity
+	media   *fakeMediaResolver
 	dedup   *fakeDedup
 	binder  *fakeBinder
 	audit   *fakeAuditor
@@ -231,6 +247,7 @@ func newHarness(t *testing.T) *harness {
 	h := &harness{
 		inst:    &fakeInstaller{inst: activeResolved(t)},
 		ident:   &fakeIdentity{id: ResolvedIdentity{UserID: uuidFromString(t, "44444444-4444-4444-4444-444444444444")}},
+		media:   &fakeMediaResolver{},
 		dedup:   &fakeDedup{token: uuidFromString(t, "55555555-5555-5555-5555-555555555555")},
 		binder:  &fakeBinder{ensureID: uuidFromString(t, "66666666-6666-6666-6666-666666666666"), appendResult: AppendResult{DedupMarked: true}},
 		audit:   &fakeAuditor{},
@@ -244,6 +261,7 @@ func newHarness(t *testing.T) *harness {
 	h.router.Register(channel.TypeFeishu, ResolverSet{
 		Installation: h.inst,
 		Identity:     h.ident,
+		Media:        h.media,
 		Dedup:        h.dedup,
 		Session:      h.binder,
 		Audit:        h.audit,
@@ -254,6 +272,53 @@ func newHarness(t *testing.T) *harness {
 	return h
 }
 
+func TestRouterResolvesMediaOnlyAfterAccessChecks(t *testing.T) {
+	h := newHarness(t)
+	h.ident.err = ErrSenderNotMember
+	if err := h.router.Handle(context.Background(), p2pMessage(t)); err != nil {
+		t.Fatal(err)
+	}
+	if h.media.calls != 0 {
+		t.Fatalf("media resolver called %d times before membership passed", h.media.calls)
+	}
+
+	h = newHarness(t)
+	msg := p2pMessage(t)
+	msg.Source.ChatType = channel.ChatTypeGroup
+	if err := h.router.Handle(context.Background(), msg); err != nil {
+		t.Fatal(err)
+	}
+	if h.media.calls != 0 {
+		t.Fatalf("media resolver called %d times for a group message not addressed to the bot", h.media.calls)
+	}
+}
+
+func TestRouterMediaCleanupTracksAtomicAppend(t *testing.T) {
+	h := newHarness(t)
+	resolved := p2pMessage(t)
+	resolved.MediaRefs = []channel.MediaRef{{Type: channel.MsgTypeFile, StorageKey: "workspaces/ws/channel-inbound/id.pdf"}}
+	h.media.result = resolved
+	h.binder.ensureErr = errors.New("database unavailable")
+	if err := h.router.Handle(context.Background(), p2pMessage(t)); err == nil {
+		t.Fatal("expected ensure failure")
+	}
+	if h.media.cleanupCalls != 1 {
+		t.Fatalf("cleanup calls after failed transaction = %d, want 1", h.media.cleanupCalls)
+	}
+
+	h = newHarness(t)
+	h.media.result = resolved
+	if err := h.router.Handle(context.Background(), p2pMessage(t)); err != nil {
+		t.Fatal(err)
+	}
+	if h.media.cleanupCalls != 0 {
+		t.Fatalf("cleanup calls after committed append = %d, want 0", h.media.cleanupCalls)
+	}
+	if got := h.binder.lastAppend.Message.MediaRefs; len(got) != 1 || got[0].StorageKey != resolved.MediaRefs[0].StorageKey {
+		t.Fatalf("append media refs = %#v", got)
+	}
+}
+
 func TestRouter_NoResolverSet_ReturnsError(t *testing.T) {
 	h := newHarness(t)
 	msg := p2pMessage(t)