feat: add cost tracking and usage dashboard#35
feat: add cost tracking and usage dashboard#35adubuc-cloudgeometry wants to merge 1 commit intomainfrom
Conversation
Add LangWatch-powered cost tracking and usage analytics:
Backend:
- LangWatchService: fetch traces, parse metrics, aggregate per-flow usage
- Usage API router: GET /usage/, GET /usage/{flow_id}/runs, key management
- GlobalSettings model for encrypted API key storage
- Model name normalization (claude-haiku-4-5-20251001 → claude-haiku-4.5)
to match LangWatch pricing table
- Local cost fallback table (temporary until LangWatch exposes cost via API)
- Flow ID in trace metadata for rename-safe ownership matching
- Token injection for Anthropic/streaming responses
Frontend:
- UsagePage with summary cards (cost, invocations, active flows)
- LangWatchKeyForm for admin API key configuration
- LangWatchService client
- TypeScript types for usage data
Tests:
- 20 backend tests (service, API, schemas, parsing, ownership, caching)
- Frontend component and hook tests
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
WalkthroughThis pull request introduces a comprehensive Usage & Cost Tracking system integrating LangWatch analytics into LangFlow. It includes a new API router with four endpoints for querying usage summaries and flow runs, managing LangWatch API keys, and viewing key status; a full-featured LangWatchService with caching, ownership filtering, pagination, and encryption; database models for global settings storage; and frontend components for displaying usage dashboards and managing API keys. Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client (Browser)
participant API as Usage API Router
participant Service as LangWatchService
participant Redis as Redis Cache
participant LangWatch as LangWatch API
participant DB as Database
Client->>API: GET /api/v1/usage/?from_date=...&to_date=...
API->>API: Verify user (auth check)
API->>DB: Fetch flow IDs for user
DB-->>API: flow_ids
API->>Service: get_usage_summary(params, flow_ids, api_key)
Service->>Redis: Check cache with key
Redis-->>Service: Cache hit or miss
alt Cache Hit
Service-->>API: Cached UsageResponse
else Cache Miss
Service->>LangWatch: POST /api/traces/search (pagination loop)
LangWatch-->>Service: Traces with cost, tokens, models
Service->>DB: Query Flow/User metadata (ownership)
DB-->>Service: FlowMeta (flow_id, user_id, username)
Service->>Service: Filter traces by allowed flow_ids
Service->>Service: Aggregate: group by flow, compute totals/averages
Service->>Redis: Write aggregated result (setex with ttl)
Service-->>API: UsageResponse (uncached)
end
API-->>Client: HTTP 200 + UsageResponse (with flows and summary)
sequenceDiagram
participant Client as Client (Browser/Admin)
participant API as Usage API Router
participant Service as LangWatchService
participant DB as Database
Client->>API: POST /api/v1/usage/settings/langwatch-key<br/>{api_key: "..."}
API->>API: Verify admin
API->>Service: validate_key(api_key)
Service->>Service: Trim whitespace
Service->>LangWatch: POST /api/traces/search (minimal payload)
LangWatch-->>Service: 200 OK or error
Service-->>API: True or False
alt Valid Key
API->>Service: save_key(api_key, admin_user_id)
Service->>Service: Encrypt key via Fernet
Service->>DB: Save to global_settings (key, encrypted_value, updated_by)
DB-->>Service: Success
Service->>Redis: invalidate_cache()
API-->>Client: HTTP 200 + SaveKeyResponse<br/>(success=true, key_preview="****abc")
else Invalid Key
API-->>Client: HTTP 422 + ErrorDetail<br/>(code="INVALID_KEY")
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes 🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
📝 Coding Plan
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment Tip CodeRabbit can generate a title for your PR based on the changes with custom instructions.Set the |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 20
🧹 Nitpick comments (28)
langbuilder/src/backend/base/tests/services/test_langwatch_ownership.py (1)
38-56: Consider consolidating duplicate fixtures.There are two similar fixtures:
make_service(lines 43-48) which uses themock_dbfixture, andservice(lines 51-56) which creates its ownMagicMock. This duplication could be simplified.♻️ Suggested consolidation
`@pytest.fixture` -def make_service(mock_db): - svc = LangWatchService.__new__(LangWatchService) - svc._db_session = mock_db - svc._client = LangWatchService._create_httpx_client() - return svc - - -@pytest.fixture -def service(): +def service(mock_db): + """LangWatchService instance with mocked DB.""" svc = LangWatchService.__new__(LangWatchService) - svc._db_session = MagicMock() + svc._db_session = mock_db svc._client = LangWatchService._create_httpx_client() return svcThen update tests that use
make_serviceto useservicedirectly, and passmock_dbwhere needed for assertions.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/services/test_langwatch_ownership.py` around lines 38 - 56, Consolidate the duplicate fixtures by removing make_service and reusing a single service fixture that accepts the mock_db fixture: change the service fixture to set svc._db_session = mock_db (instead of creating a new MagicMock) while still creating svc via LangWatchService.__new__ and svc._client via LangWatchService._create_httpx_client(); then update tests that referenced make_service to use service and rely on the mock_db fixture for assertions.langbuilder/src/backend/base/tests/services/test_langwatch_api_spike.py (1)
241-247: Split compound assertion for clearer failure messages.The assertion at lines 245-246 combines two checks. If it fails, it's unclear which condition failed. Per static analysis hint PT018, consider splitting into separate assertions.
♻️ Suggested fix
def test_pagination_scroll_id_is_non_empty_string(self): - """scrollId in sample response must be non-empty (simulates a real paged response).""" + """ScrollId in sample response must be non-empty (simulates a real paged response).""" data = _load_fixture() scroll_id = data["pagination"]["scrollId"] - assert isinstance(scroll_id, str) and len(scroll_id) > 0, ( - "Sample fixture should have a non-null scrollId to test pagination logic" - ) + assert isinstance(scroll_id, str), ( + "Sample fixture scrollId must be a string, got: " + str(type(scroll_id)) + ) + assert len(scroll_id) > 0, ( + "Sample fixture scrollId must be non-empty to test pagination logic" + )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/services/test_langwatch_api_spike.py` around lines 241 - 247, In test_pagination_scroll_id_is_non_empty_string, split the compound assertion that checks scroll_id into two distinct assertions so failures are explicit: first assert isinstance(scroll_id, str) with a message like "scrollId must be a string", then assert len(scroll_id) > 0 (or scroll_id != "") with a message like "scrollId must be non-empty"; locate the validation around the scroll_id variable loaded via _load_fixture() and replace the single combined assert with these two separate assertions.langbuilder/src/backend/base/tests/services/test_langwatch_fetch.py (1)
114-115: Movejsonimport to module level.The
jsonimport is done inline within multiple test functions (Lines 114, 158, 193). Consider moving it to the module-level imports for consistency.♻️ Suggested fix
Add to the imports at the top of the file:
import jsonThen remove the inline
import jsonstatements from lines 114, 158, and 193.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/services/test_langwatch_fetch.py` around lines 114 - 115, Move the inline json imports to the module level: add a single "import json" to the top-level imports of the test module and remove the three inline "import json" statements currently inside the test functions (the places that call json.loads(requests[0].content) and similar). Update references in those tests (the calls to json.loads(...)) to use the module-level import so there are no duplicated inline imports.langbuilder/src/frontend/src/pages/UsagePage/components/UsageSummaryCards.tsx (2)
14-17: Currency symbol is hardcoded.The
UsageSummarytype includes acurrencyfield, but the component always displays$. For internationalization support, consider using the currency from the summary data.♻️ Suggested currency-aware formatting
<p className="text-sm text-muted-foreground">Total Cost</p> <p className="text-2xl font-bold"> - ${summary.total_cost_usd.toFixed(4)} + {new Intl.NumberFormat(undefined, { + style: 'currency', + currency: summary.currency, + minimumFractionDigits: 4, + maximumFractionDigits: 4, + }).format(summary.total_cost_usd)} </p>Apply similar changes to the "Avg Cost / Invocation" card at Line 34.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/frontend/src/pages/UsagePage/components/UsageSummaryCards.tsx` around lines 14 - 17, The component currently hardcodes "$" when rendering summary.total_cost_usd and the "Avg Cost / Invocation" value; update the rendering to use the currency from the UsageSummary (e.g., summary.currency) and format amounts with a currency-aware formatter (Intl.NumberFormat or equivalent) instead of string interpolation; locate the JSX that renders summary.total_cost_usd and the avg cost invocation card in UsageSummaryCards (references: summary.total_cost_usd, summary.currency, avg cost calculation) and replace the literal "$" and toFixed usage with a currency-formatted output that passes summary.currency to the formatter.
9-9: Consider responsive grid for smaller screens.The
grid-cols-4class creates a fixed 4-column layout that may not display well on mobile or tablet devices. Consider using responsive breakpoints.♻️ Suggested responsive layout
- <div className="grid grid-cols-4 gap-4" data-testid="usage-summary-cards"> + <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-4" data-testid="usage-summary-cards">🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/frontend/src/pages/UsagePage/components/UsageSummaryCards.tsx` at line 9, The grid currently uses a fixed "grid grid-cols-4 gap-4" which breaks on small screens; update the container in UsageSummaryCards (the div with data-testid="usage-summary-cards") to use responsive Tailwind column classes such as "grid grid-cols-1 sm:grid-cols-2 md:grid-cols-3 lg:grid-cols-4 gap-4" (or a similar breakpoint set) so the layout collapses to one column on mobile and progressively increases columns on larger screens.langbuilder/src/backend/base/tests/api/test_usage_security.py (2)
362-382: Same issue: test doesn't exercise actual endpoint behavior.Similar to the previous test,
test_langwatch_key_status_non_admin_gets_403_via_dependencypatches the dependency and raises it directly without calling the endpoint. This pattern validates Python's exception mechanism rather than the endpoint's security.The tests at lines 270-304 and 339-359 that inspect function signatures are more meaningful for verifying dependency declarations.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/api/test_usage_security.py` around lines 362 - 382, The test test_langwatch_key_status_non_admin_gets_403_via_dependency wrongly validates Python exception raising instead of exercising the endpoint; modify it to call the actual GET endpoint ("/settings/langwatch-key/status") using the FastAPI test client or AsyncClient against the app instance while patching langflow.services.auth.utils.get_current_active_superuser to raise HTTPException(403), then assert the response status_code is 403 (and optionally inspect response.json()); keep the dependency patch on get_current_active_superuser and reference the test function name to locate and update the test.
310-334: Test doesn't actually exercise the endpoint behavior.The test
test_langwatch_key_endpoint_non_admin_gets_403_via_dependencypatches the dependency and immediately raises the patched exception. This only proves the patch works, not that the endpoint correctly uses the dependency.Consider calling the actual endpoint function with a non-superuser mock to verify the dependency integration, or document that this test validates the dependency mechanism rather than the endpoint behavior.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/api/test_usage_security.py` around lines 310 - 334, The test currently only raises the patched exception directly and doesn't exercise the endpoint; update test_langwatch_key_endpoint_non_admin_gets_403_via_dependency to call the actual endpoint (save_langwatch_key route) while overriding the get_current_active_superuser dependency to raise HTTPException(403) so you assert the endpoint returns 403; use the FastAPI app's dependency_overrides or test client/AsyncClient to inject the mock for get_current_active_superuser and then perform the POST to /settings/langwatch-key and assert the response status_code is 403, referencing the test function name and the dependency get_current_active_superuser to locate the change.langbuilder/src/backend/base/tests/services/test_langwatch_schemas.py (1)
140-149: Consider using dict literal instead ofdict()call.Static analysis flagged this as a minor style issue. Using
{}is marginally faster and more idiomatic in Python.♻️ Proposed fix
def _make_summary(self, **kwargs): - defaults = dict( + defaults = { - total_cost_usd=1.23, + "total_cost_usd": 1.23, - total_invocations=100, + "total_invocations": 100, - avg_cost_per_invocation_usd=0.0123, + "avg_cost_per_invocation_usd": 0.0123, - active_flow_count=5, + "active_flow_count": 5, - date_range=DateRange(from_=date(2026, 1, 1), to=date(2026, 3, 1)), + "date_range": DateRange(from_=date(2026, 1, 1), to=date(2026, 3, 1)), - ) + } defaults.update(kwargs) return UsageSummary(**defaults)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/services/test_langwatch_schemas.py` around lines 140 - 149, Replace the dict() call in the _make_summary helper with a dict literal; specifically, change the defaults = dict(...) in the _make_summary method to use a literal like defaults = { ... } while keeping the same keys/values for total_cost_usd, total_invocations, avg_cost_per_invocation_usd, active_flow_count and date_range (which is used to construct UsageSummary(**defaults)); no behavior changes, just a stylistic swap in test_langwatch_schemas.py.langbuilder/src/backend/base/tests/services/test_langwatch_key_validation.py (1)
29-35: Test helper bypasses constructor - ensure this matches production behavior.The
_make_service()helper creates aLangWatchServiceinstance using__new__and manually sets private attributes. While this is a valid testing pattern, ensure that changes to the service's initialization logic are reflected here to avoid test drift.The private member access warnings (SLF001) from static analysis are acceptable here since this is test code that needs to mock internal state.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/services/test_langwatch_key_validation.py` around lines 29 - 35, The test helper _make_service currently constructs LangWatchService via LangWatchService.__new__ and manually sets private attributes (_db_session, _client, redis), which can drift from production initialization; change the helper to instantiate via the real constructor (call LangWatchService(...) or call svc.__init__(...) with appropriate mocked dependencies) so production init logic runs, or if you must bypass __init__, document and mirror any constructor behavior by invoking the same initialization steps (e.g., call LangWatchService.__init__(svc, db_session=AsyncMock(), client=MagicMock(spec=httpx.AsyncClient), redis=None)) and keep the helper updated when LangWatchService.__init__ changes to prevent test drift.langbuilder/src/backend/base/tests/api/test_usage_router_skeleton.py (2)
19-40: Consider cleanup of stubbed modules to prevent test pollution.The helper stubs modules in
sys.modulesbut doesn't clean them up after use. This could affect other tests if they expect different behavior from these modules. Consider using pytest fixtures with cleanup.♻️ Proposed fix using fixture with cleanup
+import pytest + +@pytest.fixture +def router_module(): + """Load the usage router module with proper cleanup.""" + stubs_added = [] + optional_stubs = [ + "langflow.api.utils", + "langflow.api.utils.core", + "fastapi_pagination", + "lfx.services.deps", + ] + for mod in optional_stubs: + if mod not in sys.modules: + sys.modules[mod] = MagicMock() + stubs_added.append(mod) + + router_path = Path(__file__).parent.parent.parent / "langflow" / "api" / "v1" / "usage" / "router.py" + spec = importlib.util.spec_from_file_location("langflow.api.v1.usage.router", router_path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + yield module + + # Cleanup stubbed modules + for mod in stubs_added: + sys.modules.pop(mod, None)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/api/test_usage_router_skeleton.py` around lines 19 - 40, The helper _load_router_module injects stubs into sys.modules via optional_stubs and hard-coded "lfx.services.deps" but never removes them, which can pollute other tests; update the test to register and remove these entries after use (or convert _load_router_module into a pytest fixture that yields the module and performs cleanup), restoring any previously existing sys.modules entries and deleting the injected keys when finished so optional_stubs and "lfx.services.deps" are not left behind.
36-36: Hardcoded relative path is fragile.The path calculation assumes a specific directory structure. Consider using a more robust approach or documenting the expected structure.
#!/bin/bash # Verify the router file exists at the expected path fd -t f "router.py" --path-separator "/" | grep -E "api/v1/usage/router.py$"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/api/test_usage_router_skeleton.py` at line 36, The test uses a fragile hardcoded Path assigned to router_path; replace it with a robust lookup using Python's import machinery (e.g., importlib.util.find_spec or importlib.resources) or resolve the installed package module file to locate the router module file dynamically; specifically, change the code that sets router_path (the variable assigned from Path(__file__).parent.parent.parent / "langflow" / "api" / "v1" / "usage" / "router.py") to compute the path via importlib.util.find_spec("langflow.api.v1.usage").origin or importlib.resources.files("langflow.api.v1.usage").joinpath("router.py") and assert existence, so the test works regardless of project layout.langbuilder/src/backend/base/tests/services/test_langwatch_exceptions.py (1)
51-53: Consider assigning exception message to a variable.Static analysis flags the string literal in the exception. While minor in test code, assigning to a variable is slightly cleaner.
♻️ Proposed fix
+ test_message = "test message" for exc_class in exception_classes: with pytest.raises(LangWatchError): - raise exc_class("test message") + raise exc_class(test_message)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/services/test_langwatch_exceptions.py` around lines 51 - 53, Assign the exception message to a variable and use that variable when raising inside the test loop to avoid a raw string literal; e.g., introduce message = "test message" before the for loop (or at the start of the loop) and replace raise exc_class("test message") with raise exc_class(message) in the test that iterates over exception_classes and expects LangWatchError via pytest.raises.langbuilder/src/frontend/src/pages/UsagePage/hooks/__tests__/useGetUsageSummary.test.ts (1)
21-66: Exercise this hook through React, not as a plain function.These cases only work because
useQueryis mocked into a regular function. CallinguseGetUsageSummary()directly bypasses hook lifecycle/context behavior, so the suite will miss regressions if the hook later adds any React logic. PreferrenderHook(() => useGetUsageSummary(params))(or the repo’s equivalent hook test wrapper) and keep asserting againstmockUseQuery.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/frontend/src/pages/UsagePage/hooks/__tests__/useGetUsageSummary.test.ts` around lines 21 - 66, The tests call the hook directly which bypasses React hook lifecycle; update each test to exercise useGetUsageSummary via a hook renderer (e.g. use the repo’s hook test wrapper or react-hooks-library’s renderHook) by calling renderHook(() => useGetUsageSummary(params)) instead of useGetUsageSummary(params), then continue asserting against mockUseQuery (mockUseQuery.mock.calls...) to verify queryKey, staleTime, gcTime, retry, placeholderData and retryDelay; ensure you pass the same params objects (e.g. { from_date: ..., to_date: ... } or {}) into renderHook to keep assertions valid.langbuilder/src/frontend/src/services/__tests__/LangWatchService.test.ts (1)
8-10: Mock the shared API controller here, notfetch.This suite is pinning
LangWatchServiceto its transport layer, so it will still pass if the service bypasses the repo-standard@/controllers/APIclient. Mock the API controller boundary and assert the calls made through it instead.As per coding guidelines, "Use the
apicontroller from@/controllers/APIfor backend API calls".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/frontend/src/services/__tests__/LangWatchService.test.ts` around lines 8 - 10, Tests currently mock global.fetch (mockFetch) which pins LangWatchService to the transport layer; instead remove the global.fetch mock and mock the shared API controller exported from '@/controllers/API' that LangWatchService uses (the exported api client) so the service is tested through its intended boundary. Replace the fetch mock with a jest mock of the module '@/controllers/API' (mock the exported api object's methods that LangWatchService calls), have those mocks return the expected responses, and update assertions to verify calls were made on api.<methodName> used by LangWatchService rather than on fetch/global.fetch.langbuilder/src/backend/base/langflow/services/database/models/global_settings.py (1)
7-27: Place this model in a settings-specific subdirectory.
GlobalSettingsis being added directly underservices/database/models/, but the repo convention for ORM models is to keep them in an appropriate child directory. Moving it now avoids baking a one-off path into the rest of the LangWatch code.As per coding guidelines, "Store database models in appropriate subdirectories under
src/backend/base/langflow/services/database/models/(api_key, flow, folder, user, etc.)".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/langflow/services/database/models/global_settings.py` around lines 7 - 27, The GlobalSettings model class should be moved out of the top-level models package into a settings-specific subpackage: create a new subdirectory (e.g., models/settings) and place the GlobalSettings class there (preserving the class name and fields); update any imports that reference GlobalSettings across the codebase to point to the new module, add an __init__.py in the settings package to re-export GlobalSettings for backward-compatible imports, and run/adjust any ORM registration or migration references that import the model so they use the new module path.langbuilder/src/backend/base/tests/services/test_langwatch_flow_runs.py (1)
43-58: This case doesn't actually coverflow_idmatching.The traces only differ by
Flow:label, and_make_trace()never emitsmetadata["flow_id"]. A regression in the rename-safeflow_idpath would still pass this test. Add one trace with a stale label but matchingflow_id, and assert it is returned.Also applies to: 143-170
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/services/test_langwatch_flow_runs.py` around lines 43 - 58, Test coverage doesn't exercise the flow_id matching path because _make_trace never sets metadata["flow_id"] and traces only vary by the "Flow:" label; update _make_trace (used in langwatch flow-run tests) to optionally emit metadata["flow_id"] and add a new test case (also update the tests around the second block at lines ~143-170) that creates two traces: one with a stale/incorrect "Flow:" label but metadata["flow_id"] equal to the target flow_id, and another with the correct "Flow:" label but different flow_id, then assert the function under test returns the trace that matches by flow_id. Ensure you reference _make_trace when adding the flow_id value and assert the returned trace is the one with matching metadata["flow_id"].langbuilder/src/backend/base/tests/services/test_langwatch_httpx_client.py (1)
78-90: Avoid asserting on httpx/httpcore private pool state.
_transport._pool._max_connectionsand_max_keepalive_connectionsare private internals. A dependency update can break this test while the configuredLimitsare still correct. Prefer asserting thelimits=argument passed intoAsyncClient, or expose the limits as a service constant. Please verify this against the current httpx public API before freezing the test around these fields.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/services/test_langwatch_httpx_client.py` around lines 78 - 90, The test is asserting on httpx/httpcore private internals (pool._max_connections), which is fragile; instead update the test to verify the public configuration by asserting the Limits value exposed by the service (e.g. LangWatchService.HTTPX_LIMITS) or the argument passed into AsyncClient in LangWatchService._create_httpx_client; if such a constant doesn’t exist, add a service-level constant (HTTPX_LIMITS) in LangWatchService and use that in _create_httpx_client, then change the test to assert LangWatchService.HTTPX_LIMITS == Limits(max_connections=20, max_keepalive_connections=10) (or equivalent check against the public AsyncClient argument) rather than reading client._transport._pool internals.langbuilder/src/backend/base/langflow/api/v1/usage/router.py (2)
82-127: Exception mapping is thorough but could be clearer with return type.The
_raise_langwatch_http_errorfunction always raises an exception (either HTTPException or re-raises the original). The function signature returnsNonewhich is technically accurate since it never returns normally, but usingNoReturnwould be more explicit.♻️ Use NoReturn type hint
+from typing import NoReturn + -def _raise_langwatch_http_error(exc: Exception) -> None: +def _raise_langwatch_http_error(exc: Exception) -> NoReturn: """Map LangWatch service exceptions to structured HTTP errors."""🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/langflow/api/v1/usage/router.py` around lines 82 - 127, The function _raise_langwatch_http_error always raises and should be annotated with typing.NoReturn instead of None; update its signature to use NoReturn and add the corresponding import (from typing import NoReturn) at the top of the module so static type checkers know this function never returns.
217-230: Tuple indexing for row data is fragile.The code accesses row elements by index (
row[1],row[2]), which could break if theselectcolumn order changes. Consider using named access or unpacking.♻️ Use named tuple unpacking
result = await db.execute(select(Flow.id, Flow.name, Flow.user_id).where(Flow.id == flow_id)) row = result.fetchone() if row is None: raise HTTPException( status_code=404, detail={ "code": "FLOW_NOT_FOUND", "message": "No usage data found for this flow in the selected period.", }, ) - flow_name: str = row[1] - flow_owner_id: UUID = row[2] + _, flow_name, flow_owner_id = row🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/langflow/api/v1/usage/router.py` around lines 217 - 230, The tuple indexing is fragile: when you get result.fetchone() for select(Flow.id, Flow.name, Flow.user_id) replace index access with named access or explicit unpacking to avoid order dependencies; for example, use row._mapping["name"] and row._mapping["user_id"] (or alias the selected columns and read by those keys) to set flow_name and flow_owner_id instead of row[1]/row[2], keeping the same null-check on result.fetchone() that raises the HTTPException when row is None.langbuilder/src/backend/base/tests/services/test_langwatch_caching.py (2)
475-478: Minor redundancy in async verification.The
test_invalidate_cache_is_asynctest duplicates the check intest_invalidate_cache_is_async_implementation(lines 504-514). Consider consolidating these tests.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/services/test_langwatch_caching.py` around lines 475 - 478, Remove the redundant async verification test: either delete the simpler test_invalidate_cache_is_async or merge its assertion into test_invalidate_cache_is_async_implementation so the coroutine check for LangWatchService.invalidate_cache is performed only once; locate the tests named test_invalidate_cache_is_async and test_invalidate_cache_is_async_implementation and ensure the remaining test still asserts inspect.iscoroutinefunction(LangWatchService.invalidate_cache) while keeping the more comprehensive implementation checks intact.
313-316: TTL argument extraction is slightly fragile.The logic to extract the TTL argument handles both positional and keyword forms, but the fallback chain is complex. This is acceptable for a test, but if the
setexsignature changes, this assertion could fail unexpectedly.♻️ Simpler TTL extraction
call_args = redis_mock.setex.call_args - # Second argument should be cache_ttl - ttl_arg = call_args[0][1] if call_args[0] else call_args[1].get("time") or call_args[0][1] + # setex(key, time, value) - time is the second positional arg + ttl_arg = call_args[0][1] assert ttl_arg == service.cache_ttl🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/services/test_langwatch_caching.py` around lines 313 - 316, The TTL extraction is fragile; replace the complex fallback with a clear args/kwargs destructure from redis_mock.setex.call_args (e.g., assign args, kwargs = redis_mock.setex.call_args), then set ttl_arg = args[1] if len(args) > 1 else kwargs.get("time") or kwargs.get("ttl") so the test explicitly handles positional vs keyword invocation and then assert ttl_arg == service.cache_ttl; reference redis_mock.setex.call_args, ttl_arg, and service.cache_ttl when locating the change.langbuilder/src/frontend/src/services/LangWatchService.ts (1)
25-36: Extract duplicated error handling into a helper function.The same error handling logic is repeated in all four functions. Extract this into a reusable helper to improve maintainability.
♻️ Extract error handler
interface ApiError extends Error { code?: string; retryable?: boolean; } const handleApiError = async (response: Response): Promise<never> => { const data = await response.json().catch(() => ({})); const detail = data?.detail; const message = (typeof detail === "object" ? detail?.message : detail) || data?.message || response.statusText || "Unknown error"; const err: ApiError = new Error(message); err.code = typeof detail === "object" ? detail?.code : undefined; err.retryable = typeof detail === "object" ? detail?.retryable : undefined; throw err; };Then use it in each function:
if (!response.ok) { await handleApiError(response); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/frontend/src/services/LangWatchService.ts` around lines 25 - 36, Extract the duplicated error parsing/throwing logic into a single helper: add an ApiError interface and a handleApiError(response: Response): Promise<never> function that reads response.json (fallback to {}), computes detail, message, sets code and retryable on the Error, and throws it; then replace each repeated block in LangWatchService functions with "if (!response.ok) await handleApiError(response);" so all four call the shared handleApiError helper (refer to handleApiError, ApiError, and the existing response.ok checks to locate and update the code).langbuilder/src/backend/base/langflow/services/langwatch/service.py (5)
73-117: Static pricing table is a documented known limitation.The hardcoded model costs will become stale as providers update pricing. This is acknowledged in the PR as a known limitation. Consider adding a TODO comment or log warning when a model isn't found in the table to make staleness more visible.
♻️ Add warning for unknown models
def _estimate_cost(model: str | None, prompt_tokens: int | None, completion_tokens: int | None) -> float: """Estimate cost from model + tokens using LangWatch's published rates.""" if not model or prompt_tokens is None: return 0.0 rates = _MODEL_COST_PER_TOKEN.get(model) if not rates: + logger.debug("Unknown model for cost estimation: %s", model) return 0.0 return (prompt_tokens * rates[0]) + ((completion_tokens or 0) * rates[1])🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/langflow/services/langwatch/service.py` around lines 73 - 117, The hardcoded _MODEL_COST_PER_TOKEN table can become stale; update _estimate_cost to log or warn when a model lookup fails and add a TODO near _MODEL_COST_PER_TOKEN noting this is temporary and should be loaded from a live source; specifically, in _estimate_cost (and/or caller) detect when rates is None and emit a warning (use the module logger or a provided logger) including the model name so missing models are visible, and add a short TODO comment above _MODEL_COST_PER_TOKEN indicating the table is a fallback and should be replaced with a dynamic pricing fetch.
39-39: Consider movingAsyncGeneratorimport to TYPE_CHECKING block.The
AsyncGeneratorimport is only used in type hints. Moving it to theTYPE_CHECKINGblock would slightly optimize runtime imports.♻️ Move import to TYPE_CHECKING
-from collections.abc import AsyncGenerator from dataclasses import dataclass ... if TYPE_CHECKING: + from collections.abc import AsyncGenerator from redis.asyncio import Redis🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/langflow/services/langwatch/service.py` at line 39, Move the runtime import of AsyncGenerator into a TYPE_CHECKING guard: add "from typing import TYPE_CHECKING" at the top, remove the top-level "from collections.abc import AsyncGenerator", and inside "if TYPE_CHECKING:" add "from collections.abc import AsyncGenerator". Ensure any type annotations that reference AsyncGenerator still work (use forward-reference strings if necessary) so functions/classes like the async generator return types in service.py keep their type hints but avoid importing AsyncGenerator at runtime.
453-464: Complex collision handling could be simplified.The nested if logic for handling flow name collisions is hard to follow. Consider restructuring for clarity.
♻️ Simplify collision logic
- if new_allowed and not old_allowed: - flow_name_map[row.name] = meta - elif new_allowed and old_allowed: - # Both allowed (admin view) — prefer most recently created. - if hasattr(row, "created_at") and row.created_at and ( - not hasattr(existing, "created_at") - or not getattr(existing, "created_at", None) - or row.created_at > existing.created_at - ): - flow_name_map[row.name] = meta + # Prefer new if: (1) new is allowed but old isn't, or + # (2) both allowed and new is more recent + prefer_new = ( + (new_allowed and not old_allowed) or + (new_allowed and old_allowed and + getattr(row, "created_at", None) and + getattr(row, "created_at") > getattr(existing, "created_at", None)) + ) + if prefer_new: + flow_name_map[row.name] = meta🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/langflow/services/langwatch/service.py` around lines 453 - 464, The collision logic around flow_name_map assignment is nested and hard to follow; simplify by first checking new_allowed (row.id in allowed_flow_ids) and then inside that branch handle the two cases: if not old_allowed (existing.flow_id not in allowed_flow_ids) set flow_name_map[row.name] = meta, else both allowed so compare creation times (use hasattr/getattr on row.created_at and existing.created_at, treating missing/None as older) and if row.created_at is newer assign flow_name_map[row.name] = meta; reference the variables row, existing, allowed_flow_ids, flow_name_map, created_at, and meta when updating the code.
123-136: Key derivation is functional but could be stronger.Using SHA-256 to derive the Fernet key from SECRET_KEY works but isn't the strongest approach. PBKDF2 or scrypt with a salt would be more robust against brute-force attacks. However, since this matches the "existing Variable model encryption pattern" and the SECRET_KEY should already be cryptographically strong, this is acceptable.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/langflow/services/langwatch/service.py` around lines 123 - 136, The _get_fernet function currently derives the Fernet key using a direct SHA-256 of SECRET_KEY; to harden key derivation replace that with a proper KDF (e.g., PBKDF2HMAC or scrypt) using a salt and sufficient iterations and derive a 32-byte key before base64.urlsafe_b64encode—update the function _get_fernet to read or generate a salt (preferably from settings_service or a configurable constant), use PBKDF2HMAC with at least tens of thousands of iterations and the SECRET_KEY as the password to derive the 32-byte key, then return Fernet(urlsafe_b64encoded_key); if you must preserve compatibility with the Variable model pattern, implement the KDF behind a toggle or migration path and keep the existing SHA-256 derivation as a clear fallback within _get_fernet.
1048-1057: Redis integration TODO is properly documented.The comment clearly explains that Redis caching is currently non-functional because
get_redis_clientdoesn't exist. The service gracefully degrades without Redis. Consider tracking this as a follow-up issue.Would you like me to create an issue to track implementing the Redis client integration?
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/langflow/services/langwatch/service.py` around lines 1048 - 1057, The current try/except block imports and calls get_redis_client leaving redis_client as None if the symbol is missing, which is documented but untracked; implement a proper Redis integration by either providing get_redis_client (or an equivalent factory) and wiring its return into redis_client (and the service instance, e.g., self.redis) or replace this block with a pluggable cache backend factory used by the service; ensure the implementation removes the broad ImportError/AttributeError swallow, surfaces initialization errors, and keeps compatibility with the cache-aside logic in get_usage_summary that expects a possibly None self.redis.langbuilder/src/backend/base/tests/api/test_flow_runs_endpoint.py (1)
21-40: Unconventional module loading pattern for testing.The
_load_routerfunction dynamically loads the router module with stubbed dependencies. While this works, it's fragile and tightly coupled to the file structure. Consider using standard FastAPI testing patterns withTestClientif possible.However, if the full app context isn't available in this test environment, this approach is acceptable as a pragmatic solution.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@langbuilder/src/backend/base/tests/api/test_flow_runs_endpoint.py` around lines 21 - 40, The test uses a fragile, filesystem-dependent dynamic import in _load_router with helper _stub_modules that injects MagicMock entries into sys.modules and constructs router_path/spec to exec_module; replace this with a more robust FastAPI testing pattern by importing the application/router directly (or creating the router via the package API) and using TestClient to exercise endpoints, or—if the app context truly cannot be imported—encapsulate the stubbing in a reusable fixture that clearly documents and isolates the sys.modules injections (keep _stub_modules but move it to a fixture and avoid hardcoded Path construction and exec_module calls) so tests no longer rely on file paths and manual exec_module behavior.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: a30561d2-85dc-4a34-af39-dd151ff59547
📒 Files selected for processing (41)
langbuilder/src/backend/base/langflow/api/v1/usage/__init__.pylangbuilder/src/backend/base/langflow/api/v1/usage/router.pylangbuilder/src/backend/base/langflow/services/database/models/global_settings.pylangbuilder/src/backend/base/langflow/services/langwatch/__init__.pylangbuilder/src/backend/base/langflow/services/langwatch/exceptions.pylangbuilder/src/backend/base/langflow/services/langwatch/schemas.pylangbuilder/src/backend/base/langflow/services/langwatch/service.pylangbuilder/src/backend/base/langflow/services/tracing/langwatch.pylangbuilder/src/backend/base/tests/api/test_flow_runs_endpoint.pylangbuilder/src/backend/base/tests/api/test_langwatch_key_endpoint.pylangbuilder/src/backend/base/tests/api/test_usage_api_integration.pylangbuilder/src/backend/base/tests/api/test_usage_endpoint.pylangbuilder/src/backend/base/tests/api/test_usage_router_registration.pylangbuilder/src/backend/base/tests/api/test_usage_router_skeleton.pylangbuilder/src/backend/base/tests/api/test_usage_security.pylangbuilder/src/backend/base/tests/services/test_langwatch_api_spike.pylangbuilder/src/backend/base/tests/services/test_langwatch_caching.pylangbuilder/src/backend/base/tests/services/test_langwatch_encryption.pylangbuilder/src/backend/base/tests/services/test_langwatch_exceptions.pylangbuilder/src/backend/base/tests/services/test_langwatch_fetch.pylangbuilder/src/backend/base/tests/services/test_langwatch_flow_runs.pylangbuilder/src/backend/base/tests/services/test_langwatch_httpx_client.pylangbuilder/src/backend/base/tests/services/test_langwatch_key_validation.pylangbuilder/src/backend/base/tests/services/test_langwatch_ownership.pylangbuilder/src/backend/base/tests/services/test_langwatch_parsing.pylangbuilder/src/backend/base/tests/services/test_langwatch_schemas.pylangbuilder/src/backend/base/tests/services/test_langwatch_service_integration.pylangbuilder/src/backend/base/tests/services/test_langwatch_service_skeleton.pylangbuilder/src/frontend/src/pages/SettingsPage/LangWatchKeyForm.tsxlangbuilder/src/frontend/src/pages/SettingsPage/__tests__/LangWatchKeyForm.test.tsxlangbuilder/src/frontend/src/pages/SettingsPage/__tests__/LangWatchKeyFormComprehensive.test.tsxlangbuilder/src/frontend/src/pages/UsagePage/UsagePage.tsxlangbuilder/src/frontend/src/pages/UsagePage/__tests__/UsagePage.test.tsxlangbuilder/src/frontend/src/pages/UsagePage/components/UsageSummaryCards.tsxlangbuilder/src/frontend/src/pages/UsagePage/components/__tests__/UsageSummaryCards.test.tsxlangbuilder/src/frontend/src/pages/UsagePage/hooks/__tests__/useGetUsageSummary.test.tslangbuilder/src/frontend/src/pages/UsagePage/hooks/useGetUsageSummary.tslangbuilder/src/frontend/src/services/LangWatchService.tslangbuilder/src/frontend/src/services/__tests__/LangWatchService.test.tslangbuilder/src/frontend/src/types/__tests__/usage.test.tslangbuilder/src/frontend/src/types/usage.ts
| created_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc)) | ||
| updated_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc)) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n langbuilder/src/backend/base/langflow/services/database/models/global_settings.pyRepository: CloudGeometry/langbuilder
Length of output: 1305
🏁 Script executed:
# Find all SQLModel/Pydantic models with updated_at fields to see existing patterns
rg "updated_at.*Field" langbuilder/src/backend/base/langflow/services/database/models/ -A 1Repository: CloudGeometry/langbuilder
Length of output: 2287
🏁 Script executed:
# Check for imports in the file to understand dependencies
head -30 langbuilder/src/backend/base/langflow/services/database/models/global_settings.pyRepository: CloudGeometry/langbuilder
Length of output: 1116
🏁 Script executed:
# Search for any onupdate patterns in the codebase
rg "onupdate" langbuilder/src/backend/base/langflow/services/database/models/ -A 2 -B 2Repository: CloudGeometry/langbuilder
Length of output: 51
🏁 Script executed:
# Look for sa_column_kwargs usage patterns
rg "sa_column_kwargs" langbuilder/src/backend/ -A 2 -B 2Repository: CloudGeometry/langbuilder
Length of output: 51
🏁 Script executed:
# Check pyproject.toml or requirements for SQLModel version
fd -e "pyproject.toml" -o -e "requirements*.txt" langbuilder/src/backend | head -5 | xargs rg "sqlmodel|pydantic" -iRepository: CloudGeometry/langbuilder
Length of output: 50517
🏁 Script executed:
grep -A 5 -B 5 "sqlmodel" langbuilder/pyproject.tomlRepository: CloudGeometry/langbuilder
Length of output: 51
🏁 Script executed:
# Check if there's any service code that manually updates updated_at fields
rg "updated_at\s*=" langbuilder/src/backend/base/langflow/services/database/ -t py --context 3Repository: CloudGeometry/langbuilder
Length of output: 822
🏁 Script executed:
# Look for any update mechanisms in the services layer
rg "update.*global_settings" langbuilder/src/backend/ -i -t py --context 2Repository: CloudGeometry/langbuilder
Length of output: 51
🏁 Script executed:
# Look for the broader context of how updates are handled
rg "flag_modified" langbuilder/src/backend/base/langflow/services/database/ -t py --context 2Repository: CloudGeometry/langbuilder
Length of output: 1145
🏁 Script executed:
# Check for any Column definitions with onupdate
rg "Column|onupdate" langbuilder/src/backend/base/langflow/services/database/models/ -t py --context 1Repository: CloudGeometry/langbuilder
Length of output: 15478
🏁 Script executed:
# Search for how GlobalSettings is updated (if at all)
rg "GlobalSettings" langbuilder/src/backend/ -t py --context 3 | head -80Repository: CloudGeometry/langbuilder
Length of output: 7887
🏁 Script executed:
# Look for pyproject.toml with broader search
find langbuilder -name "pyproject.toml" -exec grep -l "sqlmodel" {} \;Repository: CloudGeometry/langbuilder
Length of output: 112
🏁 Script executed:
cat langbuilder/src/backend/base/pyproject.toml | grep -A 10 -B 10 "sqlmodel\|sqlalchemy"Repository: CloudGeometry/langbuilder
Length of output: 2005
🏁 Script executed:
# Check the langwatch service to see how GlobalSettings updates are handled
grep -A 20 "updated_by = admin_user_id" langbuilder/src/backend/base/langflow/services/langwatch/service.pyRepository: CloudGeometry/langbuilder
Length of output: 817
🏁 Script executed:
# Look for any trigger or event listener patterns for updated_at
rg "event.listen|@event|onupdate" langbuilder/src/backend/base/langflow/services/database/ -t pyRepository: CloudGeometry/langbuilder
Length of output: 334
🏁 Script executed:
# Check if there's any CRUD operations or update service for GlobalSettings
find langbuilder/src/backend/base/langflow/services -name "*global_settings*" -o -name "*crud*" | head -20Repository: CloudGeometry/langbuilder
Length of output: 705
🏁 Script executed:
# Look at the langwatch service more thoroughly to see full update pattern
sed -n '80,150p' langbuilder/src/backend/base/langflow/services/langwatch/service.pyRepository: CloudGeometry/langbuilder
Length of output: 2636
🏁 Script executed:
# Verify if sa_column_kwargs is documented in SQLModel
# and check if onupdate works with sa_column or sa_column_kwargs
web_search SQLModel sa_column_kwargs onupdate datetime auto-updateRepository: CloudGeometry/langbuilder
Length of output: 119
🌐 Web query:
SQLModel sa_column_kwargs onupdate datetime auto-update SQLAlchemy
💡 Result:
To implement an auto-updating datetime field (e.g., updated_at) in SQLModel using sa_column_kwargs with onupdate, use the Field constructor to pass SQLAlchemy Column keyword arguments directly. This allows access to SQLAlchemy's onupdate feature without creating a full Column object via sa_column, avoiding issues like column sharing in inheritance. Example for an updated_at field that auto-updates on record changes: from datetime import datetime, timezone from typing import Optional from sqlalchemy import Column, DateTime from sqlmodel import Field, SQLModel class Item(SQLModel, table=True): id: Optional[int] = Field(default=None, primary_key=True) updated_at: Optional[datetime] = Field( default_factory=lambda: datetime.now(timezone.utc), sa_type=DateTime(timezone=True), sa_column_kwargs={"onupdate": lambda: datetime.now(timezone.utc)} ) For created_at (set only on insert), use default_factory without onupdate: created_at: Optional[datetime] = Field( default_factory=lambda: datetime.now(timezone.utc), sa_type=DateTime(timezone=True) ) This approach works well in base classes for inheritance across multiple tables, as shown in GitHub discussions solving Column assignment errors. Prefer server_onupdate=text("CURRENT_TIMESTAMP") or func.now for database-side updates to avoid Python-evaluated timestamps differing from DB transaction time: from sqlalchemy import text, func sa_column_kwargs={"onupdate": func.now} Note: Official SQLModel docs do not explicitly document sa_column_kwargs (focus on sa_column/sa_type), but it is used in SQLModel source/tests and community examples for advanced SQLAlchemy integration. For full control, use sa_column=Column(DateTime(timezone=True), onupdate=func.now).
Citations:
- 1: https://medium.com/@jtgraham38/how-to-make-auto-updating-timestamp-fields-in-sqlmodel-03c1d674fa99
- 2: https://jacob-t-graham.com/2024/03/01/how-to-make-auto-updating-timestamp-fields-in-sqlmodel/
- 3: How to refactor columns that needs sa_solumn attribute across multiple SQLModel tables? fastapi/sqlmodel#954
- 4: Many-to-many model inherits the parent class model. The use of updated_at and created_at in the model reports errors about sa_column. fastapi/sqlmodel#876
- 5: 📝 Add code example for
sa_columnonupdatetimestamps fastapi/sqlmodel#372 - 6: https://app.studyraid.com/en/read/11976/382188/field-types-and-attributes
- 7: https://sqlmodel.tiangolo.com/features
🏁 Script executed:
# Check the user model's CRUD to see how they handle updated_at updates
cat langbuilder/src/backend/base/langflow/services/database/models/user/crud.py | grep -A 10 "updated_at"Repository: CloudGeometry/langbuilder
Length of output: 387
🏁 Script executed:
# Look for any similar patterns in the codebase where onupdate or server-side updates are used
rg "server_default|func.now()" langbuilder/src/backend/base/langflow/services/database/models/ -B 2 -A 2Repository: CloudGeometry/langbuilder
Length of output: 1867
updated_at won't auto-update when an existing GlobalSettings record is modified.
Both timestamps use default_factory only, so they're set once at insert time. When a setting is updated (e.g., during langwatch/service.py:save_key()), the updated_at field remains stale unless every caller explicitly sets it. Add an onupdate hook to the model instead of relying on service code to remember this.
Proposed fix
- updated_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc))
+ updated_at: datetime = Field(
+ default_factory=lambda: datetime.now(tz=timezone.utc),
+ sa_column_kwargs={"onupdate": lambda: datetime.now(tz=timezone.utc)},
+ )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@langbuilder/src/backend/base/langflow/services/database/models/global_settings.py`
around lines 21 - 22, The GlobalSettings model's updated_at uses only
default_factory so it never changes on updates; modify the model
(GlobalSettings) to add an onupdate hook for updated_at (e.g., set onupdate to a
callable that returns datetime.now(tz=timezone.utc)) so updated_at is refreshed
automatically on record updates, leaving created_at as-is; update the Field
definition for updated_at to include onupdate and ensure timezone-aware datetime
is used consistently.
| if metadata and "flow_name" in metadata: | ||
| self.trace.update(metadata=(self.trace.metadata or {}) | {"labels": [f"Flow: {metadata['flow_name']}"]}) | ||
| self.trace.update(metadata=(self.trace.metadata or {}) | { | ||
| "labels": [f"Flow: {metadata['flow_name']}"], | ||
| "flow_id": self.flow_id, # Stable identifier — survives flow renames | ||
| }) |
There was a problem hiding this comment.
Always persist flow_id, not only when flow_name is present.
self.flow_id is available for every trace, but this branch skips writing it unless metadata["flow_name"] exists. Any caller that omits flow_name will still produce traces that cannot be matched after a rename, which is the exact fallback this PR is adding.
Proposed fix
- if metadata and "flow_name" in metadata:
- self.trace.update(metadata=(self.trace.metadata or {}) | {
- "labels": [f"Flow: {metadata['flow_name']}"],
- "flow_id": self.flow_id, # Stable identifier — survives flow renames
- })
+ trace_metadata: dict[str, Any] = {"flow_id": self.flow_id}
+ if metadata and "flow_name" in metadata:
+ trace_metadata["labels"] = [f"Flow: {metadata['flow_name']}"]
+
+ self.trace.update(metadata=(self.trace.metadata or {}) | trace_metadata)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@langbuilder/src/backend/base/langflow/services/tracing/langwatch.py` around
lines 164 - 168, The current branch only sets self.trace.update(...) when
metadata contains "flow_name", causing flow_id to be omitted for traces without
flow_name; change the logic so that self.trace.update always includes "flow_id":
self.flow_id (e.g., build a base_metadata = (self.trace.metadata or {}) |
{"flow_id": self.flow_id} and then, if "flow_name" in metadata, add the "labels"
entry to that dict) and call self.trace.update(metadata=base_metadata) so
flow_id is persisted for every trace while labels remain conditional on
metadata["flow_name"].
| # Strip date suffix: -YYYYMMDD (8 digits at end) | ||
| normalized = re.sub(r"-\d{8}$", "", model) | ||
| # Convert version dashes to dots: X-Y -> X.Y (single digits only) | ||
| normalized = re.sub(r"(\d)-(\d)", r"\1.\2", normalized) |
There was a problem hiding this comment.
Narrow the version rewrite to the Claude-style suffix.
re.sub(r"(\d)-(\d)", ...) rewrites every numeric dash pair. That turns dated IDs like gpt-4o-2024-08-06 into gpt-4o-2024.08.06, which breaks pricing-table lookup and falls back to $0 cost. Restrict the rewrite to the trailing model-version segment instead of applying it globally.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@langbuilder/src/backend/base/langflow/services/tracing/langwatch.py` around
lines 269 - 272, The global numeric-dash rewrite on the variable normalized
(from model) is too broad; update the second re.sub call so it only targets the
trailing Claude-style version segment instead of every numeric dash pair — e.g.
replace re.sub(r"(\d)-(\d)", r"\1.\2", normalized) with a pattern anchored to
the end and allowing multi-digit groups such as re.sub(r"(\d+)-(\d+)$",
r"\1.\2", normalized) so only the final version suffix is rewritten.
| def _stub_modules() -> None: | ||
| stubs = [ | ||
| "fastapi_pagination", | ||
| "langflow.api.utils", | ||
| "langflow.api.utils.core", | ||
| "lfx.services.deps", | ||
| "openai", | ||
| ] | ||
| for mod in stubs: | ||
| if mod not in sys.modules: | ||
| sys.modules[mod] = MagicMock() | ||
|
|
||
|
|
||
| def _load_router(): | ||
| _stub_modules() | ||
| router_path = Path(__file__).parent.parent.parent / "langflow" / "api" / "v1" / "usage" / "router.py" | ||
| spec = importlib.util.spec_from_file_location("langflow.api.v1.usage.router_t5", router_path) | ||
| mod = importlib.util.module_from_spec(spec) | ||
| spec.loader.exec_module(mod) | ||
| return mod |
There was a problem hiding this comment.
Don't leave _stub_modules() side effects behind.
This helper mutates sys.modules globally and never rolls it back, so one test can change imports for the rest of the run. Wrap the load in patch.dict or monkeypatch so the mocked modules disappear after each case.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@langbuilder/src/backend/base/tests/api/test_langwatch_key_endpoint.py` around
lines 20 - 39, The test helper _stub_modules mutates sys.modules globally;
change _load_router to apply the stubs inside a context that is reverted (e.g.,
use unittest.mock.patch.dict(sys.modules, {...}) or pytest's
monkeypatch.context()) so the temporary MagicMock entries are only present
during importlib loading and are removed afterward; specifically, create the
dict of stub modules (matching the names in _stub_modules) and wrap the
spec.loader.exec_module(mod) call in the patch.dict/monkeypatch context, or
replace _stub_modules with a helper that returns that dict and use it inside the
scoped patch in _load_router.
| mod = _load_router() | ||
| admin = _make_admin() | ||
| langwatch = _make_langwatch_svc() | ||
| body = SaveLangWatchKeyRequest(api_key="lw_live_abc123xyz") |
There was a problem hiding this comment.
Use non-key-looking literals in these request bodies.
These lw_live_... strings are test data, but they still match the provider's secret shape closely enough to trip gitleaks. Swap them for clearly fake placeholders or construct them from fragments.
Also applies to: 163-163
🧰 Tools
🪛 Gitleaks (8.30.0)
[high] 67-67: Detected a Generic API Key, potentially exposing access to various services and sensitive operations.
(generic-api-key)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@langbuilder/src/backend/base/tests/api/test_langwatch_key_endpoint.py` at
line 67, Replace the realistic-looking secret literal passed to
SaveLangWatchKeyRequest(api_key="lw_live_abc123xyz") with a clearly fake,
non-key-looking value (e.g. "test-api-key" or build it from harmless fragments
like "lw_" + "test" + "_key") so it won't match secret patterns; update the same
change at the other occurrence referenced (line ~163) and ensure the change is
applied where SaveLangWatchKeyRequest or any similar test request constructs an
api_key.
| response = await service_no_redis.fetch_flow_runs( | ||
| flow_id=FLOW_UUID_C, | ||
| flow_name="Other Bot", | ||
| query=FlowRunsQueryParams( | ||
| from_date=date(2026, 1, 1), | ||
| to_date=date(2026, 1, 31), | ||
| limit=10, | ||
| ), | ||
| api_key=API_KEY, | ||
| requesting_user_id=requesting_user, | ||
| is_admin=False, | ||
| ) | ||
|
|
||
| # Non-admin blocked: empty result | ||
| assert isinstance(response, FlowRunsResponse) | ||
| assert response.runs == [] | ||
| assert response.total_runs_in_period == 0 |
There was a problem hiding this comment.
Test calls fetch_flow_runs with parameters it doesn't accept.
According to the service implementation and test_flow_runs_endpoint.py (which explicitly verifies that is_admin and requesting_user_id are NOT passed to the service), ownership is enforced by the router, not the service.
This test passes requesting_user_id and is_admin to fetch_flow_runs, but the service method doesn't accept these parameters. This will cause a TypeError when the test runs.
🐛 Fix the test to match the service signature
- response = await service_no_redis.fetch_flow_runs(
- flow_id=FLOW_UUID_C,
- flow_name="Other Bot",
- query=FlowRunsQueryParams(
- from_date=date(2026, 1, 1),
- to_date=date(2026, 1, 31),
- limit=10,
- ),
- api_key=API_KEY,
- requesting_user_id=requesting_user,
- is_admin=False,
- )
+ # NOTE: Ownership is enforced by the router, not the service.
+ # This test should verify the router's behavior, not the service's.
+ # The service's fetch_flow_runs doesn't accept ownership parameters.
+ response = await service_no_redis.fetch_flow_runs(
+ flow_id=FLOW_UUID_C,
+ flow_name="Other Bot",
+ query=FlowRunsQueryParams(
+ from_date=date(2026, 1, 1),
+ to_date=date(2026, 1, 31),
+ limit=10,
+ ),
+ api_key=API_KEY,
+ )However, this test's purpose (verifying ownership blocking) should be moved to the router tests, since that's where ownership is enforced.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@langbuilder/src/backend/base/tests/services/test_langwatch_service_integration.py`
around lines 671 - 687, The test calls service_no_redis.fetch_flow_runs with
parameters it doesn't accept (requesting_user_id and is_admin); update the test
to call fetch_flow_runs only with the service-supported signature (e.g.,
flow_id/flow_name/query/api_key) by removing requesting_user_id and is_admin
from the call, and move the ownership-blocking assertion (verifying non-admin
gets empty results) into the router-level tests where ownership is enforced (so
leave this service integration test focused on service behavior only).
| """RED-phase tests for the LangWatchService skeleton (F1-T5). | ||
|
|
||
| These tests verify the public interface of LangWatchService before F2 fills in | ||
| the implementations. Every method stub is expected to raise NotImplementedError. | ||
| """ |
There was a problem hiding this comment.
These RED-phase assertions are stale now.
This PR introduces real key-management behavior on LangWatchService, but these cases still require save_key, get_stored_key, get_key_status, and validate_key to raise NotImplementedError. That either fails the suite immediately or pushes the implementation back toward stubs. Replace them with concrete behavior assertions or remove the old contract tests.
Also applies to: 82-111
🧰 Tools
🪛 Ruff (0.15.6)
[warning] 1-1: File langbuilder/src/backend/base/tests/services/test_langwatch_service_skeleton.py is part of an implicit namespace package. Add an __init__.py.
(INP001)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@langbuilder/src/backend/base/tests/services/test_langwatch_service_skeleton.py`
around lines 1 - 5, The tests in test_langwatch_service_skeleton.py are
asserting that LangWatchService methods (save_key, get_stored_key,
get_key_status, validate_key) raise NotImplementedError, but the PR implements
real key-management behavior; update the tests to assert the concrete behaviors
instead (or delete the obsolete contract tests). Specifically, replace the
NotImplementedError assertions with calls to LangWatchService.save_key followed
by assertions that LangWatchService.get_stored_key returns the saved key,
LangWatchService.get_key_status returns the expected status for that key, and
LangWatchService.validate_key returns the correct validation result (true/false)
for valid and invalid inputs; if you choose removal, delete the stale tests for
those four methods so they no longer conflict with the new implementation.
| Required to display AI cost and usage data in the Usage dashboard. | ||
| Find your API key at{" "} | ||
| <a | ||
| href="https://app.langwatch.ai" | ||
| target="_blank" | ||
| rel="noopener noreferrer" | ||
| className="underline" | ||
| > | ||
| app.langwatch.ai | ||
| </a> |
There was a problem hiding this comment.
Don't hard-code the SaaS host in this settings copy.
The backend can be pointed at a custom LANGWATCH_ENDPOINT, but this form always sends admins to app.langwatch.ai / langwatch.ai. In self-hosted or custom-endpoint deployments, the help link and billing guidance will point users away from the instance they're actually configured against.
Also applies to: 136-152
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@langbuilder/src/frontend/src/pages/SettingsPage/LangWatchKeyForm.tsx` around
lines 39 - 48, Replace the hard-coded "https://app.langwatch.ai" link/text in
the LangWatchKeyForm component with a dynamic host derived from the configured
endpoint (e.g., use the runtime config or env var used by the app such as
LANGWATCH_ENDPOINT/REACT_APP_LANGWATCH_ENDPOINT or the same config accessor the
backend uses), and update both the anchor href and visible text to reflect that
value; apply the same change for the other occurrence noted (lines ~136-152) so
the help/billing link always points to the currently configured LangWatch
instance rather than app.langwatch.ai.
| const { data, isLoading, isError, error, refetch } = useGetUsageSummary({ | ||
| from_date: debouncedDateRange.from, | ||
| to_date: debouncedDateRange.to, | ||
| user_id: userId, | ||
| sub_view: subView, | ||
| }); |
There was a problem hiding this comment.
Don't derive the admin user list from already-filtered usage data.
Once user_id is set in the query, data.flows only contains that user's rows, so uniqueUsers collapses to the current user. That makes switching directly from user A to user B impossible from the same dropdown, and a zero-result filter can leave the control empty. Keep the options from an unfiltered source (or cache the last full list) instead.
Also applies to: 78-86, 97-103
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@langbuilder/src/frontend/src/pages/UsagePage/UsagePage.tsx` around lines 37 -
42, The code is building the admin user dropdown from the already-filtered usage
response (data.flows) so when user_id is set the list collapses; change the
source for uniqueUsers to an unfiltered/full user list instead of data.flows:
either call a dedicated endpoint (e.g., useGetUsers or a new hook) or fetch
useGetUsageSummary without user_id to populate and cache the full list, and keep
using useGetUsageSummary({ from_date: debouncedDateRange.from, to_date:
debouncedDateRange.to, user_id: userId, sub_view: subView }) for results; update
the logic that computes uniqueUsers (and similar blocks around
uniqueFlows/uniqueModels) to read from the fullUsers cache/hook rather than
data.flows so the dropdown remains stable when user_id is set or when results
are empty.
| import { BASE_URL_API } from "@/constants/constants"; | ||
| import type { | ||
| FlowRunsQueryParams, | ||
| FlowRunsResponse, | ||
| KeyStatusResponse, | ||
| UsageQueryParams, | ||
| UsageResponse, | ||
| } from "@/types/usage"; | ||
|
|
||
| // Use the same base URL pattern as the rest of the app | ||
| const BASE_URL_API_V1 = BASE_URL_API; | ||
|
|
||
| export const getUsageSummary = async ( | ||
| params: UsageQueryParams, | ||
| ): Promise<UsageResponse> => { | ||
| const searchParams = new URLSearchParams(); | ||
| if (params.from_date) searchParams.set("from_date", params.from_date); | ||
| if (params.to_date) searchParams.set("to_date", params.to_date); | ||
| if (params.user_id) searchParams.set("user_id", params.user_id); | ||
| if (params.sub_view) searchParams.set("sub_view", params.sub_view); | ||
|
|
||
| const response = await fetch(`${BASE_URL_API_V1}usage/?${searchParams}`, { | ||
| credentials: "include", | ||
| }); | ||
| if (!response.ok) { | ||
| const data = await response.json().catch(() => ({})); | ||
| const detail = data?.detail; | ||
| const message = | ||
| (typeof detail === "object" ? detail?.message : detail) || | ||
| data?.message || | ||
| response.statusText || | ||
| "Unknown error"; | ||
| const err = new Error(message); | ||
| (err as any).code = typeof detail === "object" ? detail?.code : undefined; | ||
| (err as any).retryable = typeof detail === "object" ? detail?.retryable : undefined; | ||
| throw err; | ||
| } | ||
| return response.json(); | ||
| }; |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Should use the api controller instead of raw fetch.
Per coding guidelines, frontend services should use the api controller from @/controllers/API for backend API calls. Using raw fetch bypasses any centralized request/response handling, authentication, or error interceptors that the api controller might provide.
Additionally, the error enrichment pattern (err as any).code loses type safety. Consider defining a typed error class.
♻️ Suggested approach using api controller
import api from "@/controllers/API";
export const getUsageSummary = async (
params: UsageQueryParams,
): Promise<UsageResponse> => {
const searchParams = new URLSearchParams();
if (params.from_date) searchParams.set("from_date", params.from_date);
if (params.to_date) searchParams.set("to_date", params.to_date);
if (params.user_id) searchParams.set("user_id", params.user_id);
if (params.sub_view) searchParams.set("sub_view", params.sub_view);
return api.get(`usage/?${searchParams}`);
};As per coding guidelines: "Use the api controller from @/controllers/API for backend API calls".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@langbuilder/src/frontend/src/services/LangWatchService.ts` around lines 1 -
39, Replace the raw fetch in getUsageSummary with the centralized api
controller: import the default export from "@/controllers/API" and call
api.get(`usage/?${searchParams}`) from inside getUsageSummary instead of using
BASE_URL_API_V1 and fetch; remove the manual response.ok branching and JSON
error enrichment there so the api controller handles auth/interceptors and error
normalization. Also stop attaching untyped properties with (err as any) in this
file — if you need to surface error.code/retryable create or reuse a typed error
class (e.g., ApiError) in the shared API layer and propagate that instead from
the controller so callers of getUsageSummary receive typed errors.
2 participants
Summary
Key changes
Cost computation fix
claude-haiku-4-5-20251001but LangWatch pricing table expectsclaude-haiku-4.5. Two regex lines in the tracer fix this for all models.canSeeCostspermission blocks API key auth). Until they fix this, we compute cost locally using the same per-token rates from theirllmModels.json.Backend
LangWatchService: fetches traces, parses metrics, aggregates per-flow usage with Redis cache-aside patternGET /usage/,GET /usage/{flow_id}/runs, key management endpointsGlobalSettingsmodel for encrypted API key storage (Fernet, derived from SECRET_KEY)Frontend
UsagePagewith summary cards (total cost, invocations, active flows)LangWatchKeyFormfor admin API key configuration in SettingsLangWatchServiceclient for API communicationKnown limitations
total_costfield (support ticket filed) — using local fallbackflow_idin old trace metadata)Test plan
pytest tests/services/test_langwatch*.py tests/api/test_usage*.py🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Tests