Feat/gemini token usage logging

README.md

@@ -10,6 +10,8 @@ It is built for end users, not just developers: flows stay simple, readable and
 - **Describe it, don't configure it** — give a node plain-language instructions and an LLM fills in its parameters at run time. Anything you set explicitly always wins over what the model infers.
 - **Built-in MCP server** — every integration node doubles as an [MCP](https://modelcontextprotocol.io) tool. Point Claude (or any MCP client) at your BlockNext server with an API key and use your connected services from chat.
 - **AI-powered nodes** — LLMs and generative AI (text, image, audio, video) as first-class building blocks, alongside integrations for the tools you already use.
+- **Async done right** — long-running AI jobs (video, music generation) are a single node. The node starts the job, polls the provider, and returns the finished result — no wait/check/loop scaffolding on your canvas.
+- **Featherweight self-hosting** — Go services on distroless images (15–32 MB each). The full stack idles around ~150 MB RAM (even less with `TASK_RUNNER_MODE=embedded`), each API sitting in single-digit megabytes. Runs comfortably on the smallest VPS.
 - **Triggers** — start flows manually, on a schedule, or from the outside via webhooks and API calls.
 - **Live execution view** — watch every task and node progress in real time over WebSocket.
 - **Credentials management** — encrypted at rest, OAuth tokens auto-refreshed, and only ever decrypted at execution time — flows never embed secrets.

apps/platform-api/internal/llm/providers/gemini/functioncalling/function_calling_provider.go

@@ -8,10 +8,8 @@ import (
 
 	"github.com/blocknextai/go-packages/httpclient"
 	"github.com/blocknextai/go-packages/json"
-	llmCache "github.com/blocknextai/platform-api/internal/llm/cache"
 	"github.com/blocknextai/platform-api/internal/llm/functioncalling"
 	"github.com/blocknextai/platform-api/internal/llm/providers/gemini"
-	geminiCache "github.com/blocknextai/platform-api/internal/llm/providers/gemini/cache"
 )
 
 type geminiProvider struct {
@@ -24,7 +22,6 @@ type geminiProvider struct {
 	timeout           time.Duration
 	maxTimeout        time.Duration
 	client            *httpclient.Client
-	cache             llmCache.Cache
 }
 
 func New(
@@ -60,7 +57,6 @@ func New(
 		timeout:           timeout,
 		maxTimeout:        maxTimeout,
 		client:            client,
-		cache:             geminiCache.New(apiKey, model, "FunctionCalling", maxTimeout),
 	}, nil
 }
 
@@ -139,6 +135,11 @@ func (f *geminiProvider) ExecuteWithContext(ctx context.Context, data []map[stri
 
 	requestBody := map[string]any{
 		"contents": contents,
+		"systemInstruction": map[string]any{
+			"parts": []map[string]any{
+				{"text": f.systemInstruction},
+			},
+		},
 		"generationConfig": map[string]any{
 			"temperature":      temperature,
 			"topK":             f.topK,
@@ -160,16 +161,6 @@ func (f *geminiProvider) ExecuteWithContext(ctx context.Context, data []map[stri
 		},
 	}
 
-	if cacheName := f.cache.Ensure(ctx, f.systemInstruction); cacheName != "" {
-		requestBody["cachedContent"] = cacheName
-	} else {
-		requestBody["systemInstruction"] = map[string]any{
-			"parts": []map[string]any{
-				{"text": f.systemInstruction},
-			},
-		}
-	}
-
 	var successResponse SuccessResponse
 	var errorResponse gemini.ErrorResponse
 	response, err := f.client.Post("").
@@ -196,6 +187,17 @@ func (f *geminiProvider) ExecuteWithContext(ctx context.Context, data []map[stri
 		return nil, functioncalling.ErrProviderRequestFailed
 	}
 
+	usage := successResponse.UsageMetadata
+	slog.Info("Gemini function calling token usage",
+		"component", "FunctionCalling",
+		"model", f.model,
+		"prompt_tokens", usage.PromptTokenCount,
+		"cached_tokens", usage.CachedContentTokenCount,
+		"candidates_tokens", usage.CandidatesTokenCount,
+		"thoughts_tokens", usage.ThoughtsTokenCount,
+		"total_tokens", usage.TotalTokenCount,
+	)
+
 	if len(successResponse.Candidates) == 0 {
 		return nil, functioncalling.ErrNoCandidates
 	}

apps/platform-api/internal/llm/providers/gemini/functioncalling/types.go

@@ -1,7 +1,16 @@
 package functioncalling
 
 type SuccessResponse struct {
-	Candidates []Candidate `json:"candidates"`
+	Candidates    []Candidate   `json:"candidates"`
+	UsageMetadata UsageMetadata `json:"usageMetadata"`
+}
+
+type UsageMetadata struct {
+	PromptTokenCount        int32 `json:"promptTokenCount"`
+	CachedContentTokenCount int32 `json:"cachedContentTokenCount"`
+	CandidatesTokenCount    int32 `json:"candidatesTokenCount"`
+	ThoughtsTokenCount      int32 `json:"thoughtsTokenCount"`
+	TotalTokenCount         int32 `json:"totalTokenCount"`
 }
 
 type Candidate struct {

apps/platform-api/internal/llm/providers/gemini/streamingchat/streaming_provider.go

@@ -22,6 +22,7 @@ var (
 
 type geminiStreamingProvider struct {
 	client          *httpclient.Client
+	model           string
 	temperature     float32
 	topK            int32
 	topP            float32
@@ -54,6 +55,7 @@ func New(
 
 	return &geminiStreamingProvider{
 		client:          client,
+		model:           model,
 		temperature:     temperature,
 		topK:            topK,
 		topP:            topP,
@@ -116,8 +118,26 @@ func (p *geminiStreamingProvider) StreamChat(ctx context.Context, systemInstruct
 	}
 
 	ch := make(chan streamingchatPkg.Chunk, 64)
+	cachedContentUsed := payload.CachedContent != ""
 
 	go func() {
+		var usage *UsageMetadata
+		defer func() {
+			if usage == nil {
+				return
+			}
+			slog.Info("Gemini streaming token usage",
+				"component", "Generation",
+				"model", p.model,
+				"cached_content_used", cachedContentUsed,
+				"prompt_tokens", usage.PromptTokenCount,
+				"cached_tokens", usage.CachedContentTokenCount,
+				"candidates_tokens", usage.CandidatesTokenCount,
+				"thoughts_tokens", usage.ThoughtsTokenCount,
+				"total_tokens", usage.TotalTokenCount,
+			)
+		}()
+
 		defer close(ch)
 		defer func() {
 			if err := response.BodyReader.Close(); err != nil {
@@ -152,6 +172,10 @@ func (p *geminiStreamingProvider) StreamChat(ctx context.Context, systemInstruct
 				continue
 			}
 
+			if sseResp.UsageMetadata != nil {
+				usage = sseResp.UsageMetadata
+			}
+
 			for _, candidate := range sseResp.Candidates {
 				for _, p := range candidate.Content.Parts {
 					if p.Text != "" {

apps/platform-api/internal/llm/providers/gemini/streamingchat/types.go

@@ -28,7 +28,16 @@ type GenerationConfig struct {
 }
 
 type SSEResponse struct {
-	Candidates []Candidate `json:"candidates"`
+	Candidates    []Candidate    `json:"candidates"`
+	UsageMetadata *UsageMetadata `json:"usageMetadata"`
+}
+
+type UsageMetadata struct {
+	PromptTokenCount        int32 `json:"promptTokenCount"`
+	CachedContentTokenCount int32 `json:"cachedContentTokenCount"`
+	CandidatesTokenCount    int32 `json:"candidatesTokenCount"`
+	ThoughtsTokenCount      int32 `json:"thoughtsTokenCount"`
+	TotalTokenCount         int32 `json:"totalTokenCount"`
 }
 
 type Candidate struct {

apps/platform-api/internal/nodeengine/README.md

@@ -41,6 +41,8 @@ A provider's `register.go` wires each action consistently: build the node, build
 
 **Executors see lists, not single records.** The executor contract is `ExecuteWithContext(ctx, credentials map[string]any, data []map[string]any) ([]map[string]any, error)` (`domain/executors/executor.go`). `data` is the list of input records (one per upstream output item); the executor loops, `validator.Parse`s each record into its typed input struct, calls the provider API, and appends one output record per input — the array-of-records shape declared by `OutputSchema`. `credentials` is keyed by credential ID (`credentials.GetCredentials(credentials, "slack_oauth2")` returns a typed accessor over the decrypted material); executors never fetch or refresh credentials themselves.
 
+**Async providers are encapsulated, not exposed.** Long-running provider jobs (video, music, image generation) never leak their async protocol onto the canvas: the node's executor starts the job and then blocks in a `wait_*.go` helper that polls the provider's operation endpoint on a fixed interval with a retry cap, honors context cancellation, and distinguishes provider-error / status-check-failure / empty-result cases before returning the finished asset URLs as the node's ordinary output (e.g. `veo/wait_veo.go`: 10s interval, 60 retries; same pattern in `piapi`'s `wait_audiogen.go` / `wait_imagegen.go`). What generic workflow tools model as a multi-node wait/check/branch/loop-back scaffold is a single node here — one of the reasons flows can stay loop-free by design.
+
 **Validation is schema-first.** `application/jsonschema.Validator[T].Parse` copies schema `default`s into missing/empty fields, validates the record against the compiled JSON schema (`santhosh-tekuri/jsonschema`), then binds it onto the input struct via `schema:"..."` field tags with tolerant coercion (string/number/bool/slice). Executors therefore start from an already-validated, defaulted, typed input.
 
 **Registration gates.** `nodes.RegisterNode` silently skips `Disabled` nodes. `functioncalling.Generate(node)` sets `Disabled: !node.GetHasNaturalLanguage()`, and `RegisterFunctionCalling` skips disabled entries — so only natural-language nodes ever reach the LLM as functions (declaration name = node ID, dots mapped to underscores; `parameters` = the marshalled `InputSchema`). `mcp.RegisterServer` filters disabled tools out of a server's `Tools`. The `system` provider shows the gates are independent: `condition`/`sleep`/`starter` register node + executor + an MCP server but no function calling.