From 69967b25c64afe6c8d08c4c6ca6a67266e47f2e1 Mon Sep 17 00:00:00 2001 From: prosolis <5590409+prosolis@users.noreply.github.com> Date: Fri, 22 May 2026 18:55:43 -0700 Subject: [PATCH] Plumb ctx through classifier and Ollama HTTP path OllamaClient.Generate/GenerateText/call now take ctx and build the HTTP request via NewRequestWithContext, so an in-flight LLM call is aborted when the parent context is cancelled (Ctrl-C). Classifier.Classify and its tier helpers take ctx too. ProcessFunc gets a ctx parameter so the poller can forward its cancellable context down to classification. Explainer.summarize manages its own 60s context since reaction-driven flow has no parent ctx to inherit. --- internal/classifier/classifier.go | 16 +++++++++------- internal/classifier/ollama.go | 22 +++++++++++++++------- internal/explainer/explainer.go | 6 +++++- internal/ingestion/poller.go | 8 +++++--- main.go | 4 ++-- 5 files changed, 36 insertions(+), 20 deletions(-) diff --git a/internal/classifier/classifier.go b/internal/classifier/classifier.go index d11c06b..04f9dcb 100644 --- a/internal/classifier/classifier.go +++ b/internal/classifier/classifier.go @@ -1,6 +1,7 @@ package classifier import ( + "context" "encoding/json" "fmt" "log/slog" @@ -38,7 +39,8 @@ func NewClassifier(ollama *OllamaClient, matrixClient *matrix.Client) *Classifie } // Classify runs the appropriate classification pipeline for a feed item. -func (c *Classifier) Classify(item *ingestion.FeedItem) (*ClassifyResult, error) { +// ctx is checked before any LLM call so shutdown aborts in-flight work. +func (c *Classifier) Classify(ctx context.Context, item *ingestion.FeedItem) (*ClassifyResult, error) { // Direct routing — no LLM call needed if item.DirectRoute != nil { slog.Debug("direct route", "guid", item.GUID, "channel", *item.DirectRoute) @@ -60,16 +62,16 @@ func (c *Classifier) Classify(item *ingestion.FeedItem) (*ClassifyResult, error) } if item.Tier == 2 { - return c.classifyTier2(item, recent) + return c.classifyTier2(ctx, item, recent) } - return c.classifyTier1(item, recent) + return c.classifyTier1(ctx, item, recent) } -func (c *Classifier) classifyTier1(item *ingestion.FeedItem, recent []storage.RecentHeadline) (*ClassifyResult, error) { +func (c *Classifier) classifyTier1(ctx context.Context, item *ingestion.FeedItem, recent []storage.RecentHeadline) (*ClassifyResult, error) { prompt := BuildTier1Prompt(item, recent) - raw, err := c.ollama.Generate(prompt) + raw, err := c.ollama.Generate(ctx, prompt) if err != nil { c.handleOllamaFailure(err) return nil, fmt.Errorf("ollama generate: %w", err) @@ -94,7 +96,7 @@ func (c *Classifier) classifyTier1(item *ingestion.FeedItem, recent []storage.Re }, nil } -func (c *Classifier) classifyTier2(item *ingestion.FeedItem, recent []storage.RecentHeadline) (*ClassifyResult, error) { +func (c *Classifier) classifyTier2(ctx context.Context, item *ingestion.FeedItem, recent []storage.RecentHeadline) (*ClassifyResult, error) { // Run keyword gating first gatingResult, reason := RunGating(item.Headline, item.Lede) @@ -115,7 +117,7 @@ func (c *Classifier) classifyTier2(item *ingestion.FeedItem, recent []storage.Re // Stage 3 (ambiguous) or Stage 1 (definite post needing routing) — send to LLM prompt := BuildTier2Prompt(item, recent) - raw, err := c.ollama.Generate(prompt) + raw, err := c.ollama.Generate(ctx, prompt) if err != nil { c.handleOllamaFailure(err) return nil, fmt.Errorf("ollama generate: %w", err) diff --git a/internal/classifier/ollama.go b/internal/classifier/ollama.go index 5816f9a..16de5c8 100644 --- a/internal/classifier/ollama.go +++ b/internal/classifier/ollama.go @@ -2,6 +2,7 @@ package classifier import ( "bytes" + "context" "encoding/json" "fmt" "io" @@ -49,9 +50,11 @@ func NewOllamaClient(cfg config.OllamaConfig) *OllamaClient { } } -// Generate sends a prompt to Ollama via the chat API and returns the raw response text. -func (o *OllamaClient) Generate(prompt string) (string, error) { - return o.call( +// Generate sends a prompt to Ollama via the chat API and returns the raw +// response text. Honors ctx cancellation — callers can abort an in-flight +// call instead of waiting for the configured timeout. +func (o *OllamaClient) Generate(ctx context.Context, prompt string) (string, error) { + return o.call(ctx, "You are a JSON-only news classification API. Respond with valid JSON only. No preamble, no explanation, no markdown.", prompt, "json", @@ -60,11 +63,11 @@ func (o *OllamaClient) Generate(prompt string) (string, error) { // GenerateText sends a system + user prompt to Ollama and returns the raw // text response. Unlike Generate, no JSON mode is forced. -func (o *OllamaClient) GenerateText(systemPrompt, userPrompt string) (string, error) { - return o.call(systemPrompt, userPrompt, "") +func (o *OllamaClient) GenerateText(ctx context.Context, systemPrompt, userPrompt string) (string, error) { + return o.call(ctx, systemPrompt, userPrompt, "") } -func (o *OllamaClient) call(systemPrompt, userPrompt, format string) (string, error) { +func (o *OllamaClient) call(ctx context.Context, systemPrompt, userPrompt, format string) (string, error) { reqBody := chatRequest{ Model: o.model, Messages: []chatMessage{ @@ -83,7 +86,12 @@ func (o *OllamaClient) call(systemPrompt, userPrompt, format string) (string, er url := o.baseURL + "/api/chat" slog.Debug("ollama: calling", "url", url, "model", o.model) - resp, err := o.httpClient.Post(url, "application/json", bytes.NewReader(body)) + req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) + if err != nil { + return "", fmt.Errorf("ollama request build: %w", err) + } + req.Header.Set("Content-Type", "application/json") + resp, err := o.httpClient.Do(req) if err != nil { return "", fmt.Errorf("ollama request: %w", err) } diff --git a/internal/explainer/explainer.go b/internal/explainer/explainer.go index 40840d5..3a4d67c 100644 --- a/internal/explainer/explainer.go +++ b/internal/explainer/explainer.go @@ -4,6 +4,7 @@ package explainer import ( + "context" "fmt" "html" "log/slog" @@ -125,8 +126,11 @@ Rules: - Stick strictly to facts in the article. Do not speculate.` func (e *Explainer) summarize(headline, body string) (string, error) { + // Reaction-driven flow has no parent ctx; bound the LLM call ourselves. + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() prompt := fmt.Sprintf("Headline: %s\n\nArticle body:\n%s", headline, body) - out, err := e.ollama.GenerateText(summarySystem, prompt) + out, err := e.ollama.GenerateText(ctx, summarySystem, prompt) if err != nil { return "", err } diff --git a/internal/ingestion/poller.go b/internal/ingestion/poller.go index 846b632..cc79e83 100644 --- a/internal/ingestion/poller.go +++ b/internal/ingestion/poller.go @@ -12,7 +12,9 @@ import ( ) // ProcessFunc is called for each new feed item that needs classification. -type ProcessFunc func(item *FeedItem) +// ctx is the poller's parent context; callbacks should pass it into any +// downstream LLM/HTTP work so shutdown aborts in-flight calls. +type ProcessFunc func(ctx context.Context, item *FeedItem) // Poller manages per-source RSS polling goroutines. type Poller struct { @@ -182,7 +184,7 @@ func (p *Poller) pollOnceWithErr(ctx context.Context, src config.SourceConfig) e continue } - p.process(&items[i]) + p.process(ctx, &items[i]) newCount++ } @@ -212,7 +214,7 @@ func (p *Poller) pollOnceWithErr(ctx context.Context, src config.SourceConfig) e continue } - p.process(&FeedItem{ + p.process(ctx, &FeedItem{ GUID: s.GUID, Headline: s.Headline, Lede: s.Lede, diff --git a/main.go b/main.go index ee28e4a..293d2bb 100644 --- a/main.go +++ b/main.go @@ -95,8 +95,8 @@ func main() { slog.Info("post queue started") // Build the pipeline callback: classify → enqueue - processItem := func(item *ingestion.FeedItem) { - result, err := cls.Classify(item) + processItem := func(ctx context.Context, item *ingestion.FeedItem) { + result, err := cls.Classify(ctx, item) if err != nil { slog.Error("classification failed, will retry next cycle", "guid", item.GUID,