# [Agent Platform in Practice (4)] Towards Cloud Two-layer Decoupling and Full Persistence: Completing the LINE AI Business Card M

> Source: <https://dev.to/gde/agent-platform-in-practice-4-towards-cloud-two-layer-decoupling-and-full-persistence-60g>
> Published: 2026-10-23 08:32:27+00:00

In the first three articles of the "Agent Platform in Action" series, we step-by-step pushed the **LINE AI Business Card Assistant (`line-card-assistant`)** from a conceptual prototype to a cloud-native architecture:

`google.golang.org/adk/v2/workflow` to build a 12-node business card processing graph (Graph Workflow), covering Vision OCR, `AddFanOut` / `AddFanIn` parallel background investigations, and the HITL (Human-in-the-Loop) pause and modification loop of `NewRequestInputEvent`.` agents-cli` (`adk_go`) standard dual-channel architecture, allowing the project to support LINE Webhook, ADK WebUI Playground, and the A2A protocol simultaneously.`artifact/gcsartifact` to move multi-MB business card photos out of `Session.State`, and combining `session/vertexai` to implement `resolvePendingHITL`, allowing stateless containers to seamlessly continue HITL modification and saving even if scaled to 0 or instances are switched.
In today's fourth episode of the series, we will complete the **final piece of Phase 2 ([Issue #5](https://github.com/kkdai/line-card-assistant/issues/5))** and the **Phase 3 core architecture upgrade ([Issue #6](https://github.com/kkdai/line-card-assistant/issues/6))** (corresponding to **[PR #12](https://github.com/kkdai/line-card-assistant/pull/12)**):

`PostgresStore` / Cloud SQL)`HybridSearch` and token-scoring fault-tolerant `BroadSearch`.
When we deploy `line-card-assistant` to a Google Cloud production environment, a careful examination of the responsibilities of every piece of data and component in the system reveals two key problems that must be solved:

`CardStore` Business Card Database
In the previous post (PR #11), we already placed the **"original business card photos"** in **Google Cloud Storage (`GCSArtifactService`)** and the **"workflow execution state and HITL breakpoints"** in **Vertex AI / Cloud SQL (`SessionService`)**. However, when a user clicks "✅ Confirm Save" in LINE and reaches `SaveCardNode`, if the final confirmed business card data (` models.BusinessCard`) is still written to local memory `MemoryStore` or a local SQLite file inside the container:

`HybridSearchNode` won't find the card just saved in Instance B's memory!
Therefore, we must complete the final corner of the **"Cloud-Native Persistence Triad"** — **Cloud SQL PostgreSQL (`PostgresStore`)**:

| Persistence Dimension | Content Responsible for Storing | Cloud Backend Service | ADK / Go Implementation Package | 
|---|---|---|---|
| **1. Binary Large Files (Artifacts)** | Original business card photos ( `business_card.jpg` v1, v2…) | **Google Cloud Storage (GCS)** | `sessions.NewArtifactService` (`artifact/gcsartifact` ) | 
| **2. Conversation and Graph Breakpoints (Sessions)** | Graph Checkpoints, `active_interrupt_id` ,`draft_card` | **Vertex AI Agent Engine / Cloud SQL** | `sessions.NewService` (`session/vertexai` ,`session/database` ) | 
| **3. Structured Business Data (Domain DB)** | Verified `BusinessCard` official records, JSON industry tags, hybrid retrieval | **Cloud SQL for PostgreSQL** | `store.NewStore` (`internal/store/postgres.go` ) | 

In monolithic (All-in-One) mode, `cmd/server` receives LINE Webhooks, runs Vision OCR, and executes three parallel Grounding search nodes in the same process. However, in an enterprise-grade Google Cloud Agent Platform architecture, splitting the system into **"Cloud Run LINE Webhook Gateway (Edge Layer)"** and **"Google Cloud Agent Runtime (Agent Execution Layer)"** offers three major advantages:

`POST /webhook`) to push events, uses `LINE_CHANNEL_SECRET` to verify the `X-Line-Signature`, and must `200 OK` within 1-2 seconds`app.NewRootAgent` hosted on the Agent Runtime`AGENT_RUNTIME_URL` is not set, `cmd/server` automatically runs in `AGENT_RUNTIME_URL=https://...` in Cloud Run automatically switches Let's look at the cross-layer interaction under the two-tier decoupled architecture with a complete system sequence diagram:

```
sequenceDiagram
    autonumber
    actor User as LINE User
    participant GW as Cloud Run<br/>LINE Webhook Gateway<br/>(cmd/server + gateway.Client)
    participant AR as Google Cloud Agent Runtime<br/>(adkrest.Server / app.NewRootAgent)
    participant GCS as Cloud Storage<br/>(GCSArtifactService)
    participant Sess as Vertex AI / Cloud SQL<br/>(SessionService)
    participant PG as Cloud SQL PostgreSQL<br/>(PostgresStore)

    User->>GW: 1. Send business card photo (POST /webhook)
    GW-->>User: 2. Verify X-Line-Signature and immediately return HTTP 200 OK
    GW->>GW: 3. Download image from LINE Blob API []byte
    GW->>AR: 4. POST /api/apps/app/users/{user}/sessions/{id}<br/>+ POST /api/run (With Google Cloud ADC ID Token and image InlineData)
    AR->>GCS: 5. RouterNode automatically saves image to GCS Artifact ("business_card.jpg" v1)
    AR->>AR: 6. VisionOCRNode -> AddFanOut (Deduplication / Google Search company background / Conversation context) -> BuildPreviewNode
    AR->>Sess: 7. HITLConfirmNode triggers NewRequestInputEvent<br/>Persist active_interrupt_id="hitl_confirm_1"
    AR-->>GW: 8. Return Events and Session State (including flex_preview_json)
    GW-->>User: 9. Push LINE Flex Message business card preview card (⏸️ Waiting for user confirmation)

    Note over GW,AR: ☁️ Gateway is completely stateless; the next request can land on any Gateway instance

    User->>GW: 10. Input modification: "Change title to CTO" or click "✅ Confirm Save"
    GW->>AR: 11. HasPendingHITL -> GET /api/apps/app/users/{user}/sessions<br/>Automatically retrieve remote paused {SessionID, active_interrupt_id}
    GW->>AR: 12. POST /api/run (Send adk_request_input FunctionResponse)
    AR->>PG: 13. If confirming save: SaveCardNode calls PostgresStore.Save() Upsert to write to Cloud SQL
    AR->>Sess: 14. Clear active_interrupt_id="" and record saved_card_id
    AR-->>GW: 15. Return save success and AI icebreaker follow-up draft
    GW-->>User: 16. Push save confirmation and icebreaker to user!
```

`internal/store/postgres.go`) — Issue #5
In our project, since `sessions/sessions.go` already introduced `gorm.io/gorm` and `gorm.io/driver/postgres` to support `SESSION_TYPE=cloud_sql`, we can directly use GORM to build a production-grade **` PostgresStore`** for `CardStore` without extra dependency bloat!

`store.CardStore` Interface: Native While checking the original [`internal/workflow/nodes_search.go`](https://github.com/kkdai/line-card-assistant/blob/main/internal/workflow/nodes_search.go), we found a type assertion for `*store.MemoryStore` in `BroadSearchNode`:

```
// ❌ Old way: Only MemoryStore supports BroadSearch; other Stores fall back to ListByUser
if memStore, ok := cardStore.(*store.MemoryStore); ok {
    cards, err = memStore.BroadSearch(ctx, userID, keywords)
} else {
    cards, err = cardStore.ListByUser(ctx, userID)
}
```

To ensure `PostgresStore` and `MemoryStore` share a consistent hybrid retrieval and fuzzy fallback experience, we officially included `BroadSearch` in the `CardStore` standard interface in [`internal/store/store.go`](https://github.com/kkdai/line-card-assistant/blob/main/internal/store/store.go):

```
// CardStore defines the persistence and search interface for business cards.
type CardStore interface {
    Save(ctx context.Context, card *models.BusinessCard) error
    GetByID(ctx context.Context, id string) (*models.BusinessCard, error)
    CheckDuplicate(ctx context.Context, userID, name, company, email string) (*models.BusinessCard, bool, error)
    HybridSearch(ctx context.Context, filter models.SearchFilter) ([]*models.BusinessCard, error)
    BroadSearch(ctx context.Context, userID string, keywords []string) ([]*models.BusinessCard, error)
    ListByUser(ctx context.Context, userID string) ([]*models.BusinessCard, error)
}
```

This way, `BroadSearchNode` only needs to cleanly call one line: `cardStore.BroadSearch(ctx, userID, keywords)`!

`BusinessCardRecord`)
In [`internal/store/postgres.go`](https://github.com/kkdai/line-card-assistant/blob/main/internal/store/postgres.go), we defined `BusinessCardRecord` corresponding to the PostgreSQL `business_cards` table, added indexes to common query fields (`user_id`, `name`, `company`, `email`, `met_date`, `updated_at`), and stored `Tags []string` as a JSON string in `tags_json`:

```
// BusinessCardRecord is the GORM database model for persisting a models.BusinessCard
// in PostgreSQL / Cloud SQL.
type BusinessCardRecord struct {
    ID string `gorm:"primaryKey;size:128"`
    UserID string `gorm:"index;size:128;not null"`
    Name string `gorm:"index;size:255"`
    Title string `gorm:"size:255"`
    Company string `gorm:"index;size:255"`
    Email string `gorm:"index;size:255"`
    Phone string `gorm:"size:64"`
    Address string `gorm:"type:text"`
    TaxID string `gorm:"size:32"`
    Website string `gorm:"size:512"`
    RawOCR string `gorm:"type:text"`
    CompanySummary string `gorm:"type:text"`
    CompanyInfo string `gorm:"type:text"`
    TagsJSON string `gorm:"column:tags_json;type:text"`
    MetContext string `gorm:"type:text"`
    RawImageURL string `gorm:"type:text"`
    MetDate time.Time `gorm:"index"`
    CreatedAt time.Time `gorm:"index"`
    UpdatedAt time.Time `gorm:"index"`
}
```

When a user takes a photo of an existing contact's business card (e.g., the person was promoted from "Senior Engineer" to "VP of Engineering", and `CleanAndDedupNode` identifies the duplicate and reuses the original `card.ID`), `PostgresStore.Save` performs an atomic Upsert using PostgreSQL's native `ON CONFLICT (id) DO UPDATE` (`clause.OnConflict{UpdateAll: true}`):

```
if err := s.db.WithContext(ctx).Clauses(clause.OnConflict{
    Columns: []clause.Column,
    UpdateAll: true,
}).Create(rec).Error; err != nil {
    return fmt.Errorf("upsert business card %q: %w", cp.ID, err)
}
```

`store.NewStore(ctx)` Factory: Auto-sensing Cloud SQL Unix Sockets and Environment Variables
When connecting to Cloud SQL from Cloud Run or Agent Runtime, the official recommendation is to use the Unix Domain Socket (`/cloudsql/<PROJECT>:<REGION>:<INSTANCE>`) established by the Cloud SQL Auth Proxy; locally or on GKE, `DATABASE_URL` or `DB_HOST` are common. We support all three modes in `BuildPostgresDSNFromEnv()` and `store.NewStore(ctx)`, and **automatically fall back seamlessly to `MemoryStore` preloaded with 4 Taiwan sample business cards** when no database environment variables are set:

```
// BuildPostgresDSNFromEnv resolves a PostgreSQL / Cloud SQL connection string from environment variables.
// Priority:
// 1. CARD_STORE_DSN / DATABASE_URL / POSTGRES_DSN
// 2. Cloud SQL Unix socket when INSTANCE_CONNECTION_NAME + DB_PASS are set
// 3. Standard TCP host when DB_HOST + DB_PASS are set
func BuildPostgresDSNFromEnv() string {
    for _, key := range []string{"CARD_STORE_DSN", "DATABASE_URL", "POSTGRES_DSN"} {
        if v := strings.TrimSpace(os.Getenv(key)); v != "" {
            return v
        }
    }

    dbUser := envOrDefault("DB_USER", "postgres")
    dbName := envOrDefault("DB_NAME", "postgres")
    dbPass := strings.TrimSpace(os.Getenv("DB_PASS"))
    instanceConn := strings.TrimSpace(os.Getenv("INSTANCE_CONNECTION_NAME"))
    if instanceConn != "" && dbPass != "" {
        return fmt.Sprintf(
            "host=/cloudsql/%s user=%s password=%s dbname=%s sslmode=disable",
            instanceConn, dbUser, dbPass, dbName,
        )
    }
    // ...
}
```

This way, as long as your environment has `INSTANCE_CONNECTION_NAME` and `DB_PASS` set for `SESSION_TYPE=cloud_sql`, **` SessionService` and `CardStore` can share the same Cloud SQL PostgreSQL instance**, without spending an extra penny on database costs!

`AgentRuntimeClient` (`internal/gateway/client.go`) — Issue #6
Next is the core engineering of Phase 3: **How to make the `cmd/server` LINE Bot Webhook Handler forward all business card recognition, HITL recovery, search, and icebreaker requests to a remote Agent Runtime via HTTP instead of running `workflow.Engine` locally?**

Remember how we had the foresight to separate the Webhook Handler and Workflow Engine with a **`linebot.WorkflowRunner` interface** in the first version of `internal/linebot/handler.go`?

```
// internal/linebot/handler.go
type WorkflowRunner interface {
    HasPendingHITL(userID string) bool
    RecordUserMessage(userID, text string)
    StartIngest(ctx context.Context, userID string, imageBytes []byte, recentMessages []string) (*WorkflowResponse, error)
    ResumeHITL(ctx context.Context, userID string, userReply string) (*WorkflowResponse, error)
    RunSearch(ctx context.Context, userID string, query string) (*WorkflowResponse, error)
    RunEmailDraft(ctx context.Context, userID string, cardID string) (*WorkflowResponse, error)
}
```

Because of this clean interface, we **don't need to change a single line of code** in `linebot.Handler`! We just need to implement a **` gateway.Client`** that satisfies the `linebot.WorkflowRunner` interface in [`internal/gateway/client.go`](https://github.com/kkdai/line-card-assistant/blob/main/internal/gateway/client.go)!

`gateway.NewClient` Supporting Automatic Google Cloud ADC Token Issuance
When the LINE Webhook Gateway on Cloud Run needs to call an IAM-protected remote Agent Runtime (or another ADK REST Server on an internal Cloud Run), `gateway.NewClient` automatically obtains the current Service Account's Google Cloud ADC credentials via `google.golang.org/api/idtoken` (or `golang.org/x/oauth2/google`) and attaches them to the `Authorization: Bearer <token>` header of every HTTP request:

```
func NewClient(ctx context.Context, baseURL string, useADC bool, opts ...ClientOption) (*Client, error) {
    trimmed := strings.TrimRight(strings.TrimSpace(baseURL), "/")
    if trimmed == "" {
        return nil, fmt.Errorf("agent runtime baseURL cannot be empty")
    }
    if !strings.HasSuffix(trimmed, "/api") {
        trimmed += "/api"
    }

    hc := &http.Client{Timeout: 60 * time.Second}
    if useADC {
        if idClient, err := idtoken.NewClient(ctx, strings.TrimSuffix(trimmed, "/api")); err == nil {
            hc = idClient
        } else if oauthClient, err := google.DefaultClient(ctx, "https://www.googleapis.com/auth/cloud-platform"); err == nil {
            hc = oauthClient
        }
    }
    // ...
}
```

When a user uploads a business card photo on LINE:

`imageBytes` from LINE Blob API.`gateway.Client.StartIngest` calls the remote ADK REST API `POST /api/apps/{appName}/users/{userID}/sessions/{sessionID}` to create a session.`POST /api/run`, wrapping `InlineData` of `newMessage.Parts` to send to the remote Agent Runtime:

```
func (c *Client) StartIngest(ctx context.Context, userID string, imageBytes []byte, recentMessages []string) (*linebot.WorkflowResponse, error) {
    sessionID := "ingest-" + uuid.NewString()
    initialState := map[string]any{
        workflow.StateKeyInputType: workflow.RouteIngest,
        workflow.StateKeyUserID: userID,
        workflow.StateKeyRecentMessages: recentMessages,
    }
    if err := c.createSession(ctx, userID, sessionID, initialState); err != nil {
        return nil, err
    }

    parts := []*genai.Part{
        {Text: workflow.RouteIngest},
    }
    if len(imageBytes) > 0 {
        parts = append(parts, &genai.Part{
            InlineData: &genai.Blob{
                MIMEType: "image/jpeg",
                Data: imageBytes,
            },
        })
    }
    msg := &genai.Content{
        Role: genai.RoleUser,
        Parts: parts,
    }

    res, err := c.runTurn(ctx, userID, sessionID, msg, initialState)
    // ...
}
```

At this point, the logic we wrote for `RouterNode` in Phase 2 works like magic: when the remote Agent Runtime's `RouterNode` receives this request with `InlineData`, it **automatically calls `ctx.Artifacts().Save(ctx, "business_card.jpg", ...)` to store the image in the GCS Artifact Bucket**, leaving only lightweight `card_image_artifact` and `card_image_version` in `Session.State`! This means the **LINE Webhook Gateway doesn't even need direct write permissions to the GCS Bucket** for the remote Agent Runtime to complete versioned GCS Artifact archiving!

`ResumeHITL`)
The most exciting part: when a user sees the LINE Flex preview card and inputs "*Change title to CTO*" 5 minutes later, and this Webhook request lands on **another completely new Cloud Run Gateway instance**, how does `gateway.Client` know this user has a pending HITL Session on the remote Agent Runtime?

We use the `GET /api/apps/{appName}/users/{userID}/sessions` endpoint via the ADK REST API in `gateway.Client.resolvePendingHITL` to query the remote Agent Runtime for the user's session status and check `state["active_interrupt_id"]`:

```
func (c *Client) resolvePendingHITL(ctx context.Context, userID string) (pendingHITL, bool) {
    // 1. If local cache has SessionID, call remote GET /sessions/{id} to get the latest active_interrupt_id
    // (Important! Because after each Edit loop, hitl_seq increments to hitl_confirm_2, hitl_confirm_3...)
    c.mu.RLock()
    cached, hasCached := c.pendingSession[userID]
    c.mu.RUnlock()
    if hasCached && cached.SessionID != "" {
        if fullSess, err := c.getSession(ctx, userID, cached.SessionID); err == nil && fullSess != nil {
            if activeID := readStringFromMap(fullSess.State, workflow.StateKeyActiveInterruptID); activeID != "" {
                found := pendingHITL{SessionID: cached.SessionID, InterruptID: activeID}
                c.mu.Lock()
                c.pendingSession[userID] = found
                c.mu.Unlock()
                return found, true
            }
        }
    }

    // 2. If this is a brand-new stateless Gateway instance, call remote GET /sessions to find sessions with active_interrupt_id
    sessions, err := c.listSessions(ctx, userID)
    if err != nil {
        return pendingHITL{}, false
    }
    for _, sess := range sessions {
        activeID := readStringFromMap(sess.State, workflow.StateKeyActiveInterruptID)
        if activeID == "" && sess.ID != "" {
            if fullSess, err := c.getSession(ctx, userID, sess.ID); err == nil && fullSess != nil {
                activeID = readStringFromMap(fullSess.State, workflow.StateKeyActiveInterruptID)
            }
        }
        if activeID != "" {
            found := pendingHITL{SessionID: sess.ID, InterruptID: activeID}
            c.mu.Lock()
            c.pendingSession[userID] = found
            c.mu.Unlock()
            return found, true
        }
    }
    return pendingHITL{}, false
}
```

Here is a real bug and key detail we caught while writing integration tests:

`HITLConfirmNode` is `hitl_confirm_1`` ApplyEditNode` modifies the title and loops back to `hitl_confirm_2`` FunctionResponse` won't match the latest breakpoint `SessionID`, it synchronizes the latest `active_interrupt_id` from the remote via `getSession`` InterruptID`!` cmd/server/main.go`: Seamless Switching Between Monolithic and Two-Tier Gateway Modes
Finally, we added mode detection in [`cmd/server/main.go`](https://github.com/kkdai/line-card-assistant/blob/main/cmd/server/main.go):

``` js
    var runner linebot.WorkflowRunner
    if cfg.AgentRuntimeURL != "" {
        // 🌐 Mode A: Decoupled Two-Tier LINE Webhook Gateway Mode (Forwarding to remote Agent Runtime)
        gwClient, err := gateway.NewClient(
            ctx,
            cfg.AgentRuntimeURL,
            cfg.AgentRuntimeUseADC,
            gateway.WithAppName(cfg.AgentRuntimeAppName),
        )
        if err != nil {
            slog.Error("failed to initialize remote Agent Runtime gateway client", "url", cfg.AgentRuntimeURL, "err", err)
            os.Exit(1)
        }
        slog.Info("running in decoupled LINE Webhook Gateway mode",
            "agent_runtime_url", cfg.AgentRuntimeURL,
            "agent_runtime_app", cfg.AgentRuntimeAppName,
            "use_adc", cfg.AgentRuntimeUseADC,
        )
        runner = gwClient
    } else {
        // 🚀 Mode B: All-in-One Local Engine Mode (Default for local development and single container deployment)
        cardStore, err := store.NewStore(ctx)
        // ... Initialize local workflow.NewEngine ...
        runner = engine.LinebotRunner()
    }

    botHandler, err := linebot.NewHandler(cfg, runner)
```

`internal/gateway/client_test.go`)
To ensure `gateway.Client` is 100% compatible with the official ADK Go v2.6.0 `server/adkrest` REST server, we started a real **`adkrest.NewServer`** (mounting our `app.NewRootAgentWithDeps`) using `httptest.NewServer` in [`internal/gateway/client_test.go`](https://github.com/kkdai/line-card-assistant/blob/main/internal/gateway/client_test.go) and established two completely independent Gateway Clients (`client1` and `client2`) for a five-stage end-to-end test:

`client1.StartIngest`)` ArtifactService` successfully writes `business_card.jpg` $\rightarrow$ Verify return of `WaitingHITL=true` (`client2.HasPendingHITL`)` client2` probes the paused session created by `client1` via `GET /api/apps/app/users/{user}/sessions`.` client2.ResumeHITL` — Natural Language Modification)`DraftCard.Title == "CTO"`, and pausing again at `client1.ResumeHITL` — Confirm Save)`"action=confirm_save"` $\rightarrow$ `SaveCardNode` $\rightarrow$ Verify business card successfully written to remote `client2.RunSearch` & `client2.RunEmailDraft`)
Running the full test suite:

``` bash
$ go test -race -v ./...
=== RUN TestGatewayClient_EndToEndDecoupledIngestHITLAndSearch
--- PASS: TestGatewayClient_EndToEndDecoupledIngestHITLAndSearch (0.01s)
=== RUN TestBusinessCardRecordRoundTrip
--- PASS: TestBusinessCardRecordRoundTrip (0.00s)
=== RUN TestBuildPostgresDSNFromEnv
--- PASS: TestBuildPostgresDSNFromEnv (0.00s)
=== RUN TestNewStoreFactoryFallback
--- PASS: TestNewStoreFactoryFallback (0.00s)
=== RUN TestRankCardsByKeywords
--- PASS: TestRankCardsByKeywords (0.00s)
PASS
```

All unit tests and cross-layer integration tests passed with green lights under `-race` detector!

After completing **[PR #12](https://github.com/kkdai/line-card-assistant/pull/12)**, looking back at the five stages of **[Roadmap Issue #9](https://github.com/kkdai/line-card-assistant/issues/9)**, **Phase 1, Phase 2, and Phase 3 are now 100% complete**!

`agents-cli eval` Automated Evaluation and Benchmarking)
In the upcoming articles of this series, we will enter **Phase 4 and Phase 5**, using `agents-cli eval` to establish a Quality Flywheel for this 12-node Graph Workflow, and officially deploy the entire system using `agents-cli deploy` and OpenTelemetry. Stay tuned!