From 7e76fb7874fcba1d5c4911c9c4d8b8846122b147 Mon Sep 17 00:00:00 2001 From: Talha Ahmed Date: Mon, 31 Aug 2026 17:29:10 +0500 Subject: [PATCH] Find Talent: Apify spend tracking + responsive fixes - talent_runs.cost_usd records each run's actual charge (usageTotalUsd), accumulated across the broadened re-run ladder (migration 018). - GET /talent/account: live balance and cycle spend from Apify /users/me/limits (nulls when Apify is unreachable) plus the observed $/profile over recorded runs. - Header shows Balance / Spent / $-per-profile chips that refresh with every search; the run status line shows the last search's cost and the confirm dialog estimates from the live average. - Responsive: toolbar controls stack edge-to-edge and candidate card footers wrap instead of clipping their buttons on phones. Co-Authored-By: Claude Fable 5 --- backend/README.md | 6 +- .../migrations/manual/018_talent_run_cost.sql | 10 +++ backend/talent/app.py | 15 ++++ backend/talent/models.py | 60 +++++++++++--- backend/talent/plugins.py | 49 +++++++++++ backend/talent/serializers.py | 1 + backend/talent/views.py | 31 ++++++- backend/tests/test_talent_plugins.py | 51 ++++++++++++ frontend/src/api/talent.js | 25 ++++++ frontend/src/lib/queryKeys.js | 1 + frontend/src/screens/Talent.jsx | 82 ++++++++++++++++--- frontend/src/styles/styles.css | 30 +++++++ 12 files changed, 335 insertions(+), 26 deletions(-) create mode 100644 backend/migrations/manual/018_talent_run_cost.sql diff --git a/backend/README.md b/backend/README.md index 1e1ea3c..8edc020 100644 --- a/backend/README.md +++ b/backend/README.md @@ -343,7 +343,11 @@ closed tab loses nothing. Profiles are deduped per job by normalized LinkedIn UR (`is_deleted`) profile. The raw dataset item is kept verbatim in `talent_profiles.raw` because item shapes vary per actor. A run is refused with 409 while another is active for the same job, and `APIFY_MAX_COST_USD` is passed as `maxTotalChargeUsd` so Apify enforces -the spend ceiling server-side. +the spend ceiling server-side. Each run's actual charge (`usageTotalUsd`) is folded into +`talent_runs.cost_usd` when it settles (accumulating across the broadened re-run ladder), +and `GET /talent/account` serves the Find Talent header chips: live balance and +cycle spend from Apify's `/users/me/limits` (degrading to nulls when Apify is +unreachable) plus the observed $/profile over all recorded runs. ### `agent/` LangGraph state machine — see [The matching agent](#the-matching-agent). diff --git a/backend/migrations/manual/018_talent_run_cost.sql b/backend/migrations/manual/018_talent_run_cost.sql new file mode 100644 index 0000000..5413527 --- /dev/null +++ b/backend/migrations/manual/018_talent_run_cost.sql @@ -0,0 +1,10 @@ +-- 018_talent_run_cost.sql +-- Per-run Apify spend on talent_runs. `cost_usd` accumulates usageTotalUsd +-- across the re-arm ladder (one logical run can span several billed Apify +-- runs), folded in once per finished actor run. Feeds the Find Talent +-- balance/spend/$-per-profile chips. Applied at startup by +-- alembic_setup.run_manual_sql(); needed because prod boots with +-- DB_AUTOGENERATE=false. + +ALTER TABLE app.talent_runs + ADD COLUMN IF NOT EXISTS cost_usd DOUBLE PRECISION NOT NULL DEFAULT 0; diff --git a/backend/talent/app.py b/backend/talent/app.py index 60c0483..e4d3040 100644 --- a/backend/talent/app.py +++ b/backend/talent/app.py @@ -55,6 +55,21 @@ async def talent_run_status( raise HTTPException(status_code=500, detail=str(e)) +@router.get("/talent/account") +async def talent_account( + current_user: dict = Depends(require_permission(PermissionTag.TALENT_VIEW)), + session: AsyncSession = Depends(get_session), +): + try: + service = Talent(session=session) + data = await service.account() + return JSONResponse(content={"data": data, "status_code": 200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + @router.get("/talent/runs/fetch") async def fetch_talent_runs( job_post_id: str = Query(...), diff --git a/backend/talent/models.py b/backend/talent/models.py index 47a71b9..c3958bc 100644 --- a/backend/talent/models.py +++ b/backend/talent/models.py @@ -50,6 +50,10 @@ class TalentRuns(SQLModel, table=True): apify_dataset_id: str | None = Field(default=None) apify_error: str | None = Field(default=None) profiles_found: int = Field(default=0) + # Accumulated Apify spend (usageTotalUsd) across the row's whole re-arm + # ladder. server_default is load-bearing: the column arrives as an ALTER + # on a populated table (018_talent_run_cost.sql). + cost_usd: float = Field(default=0, sa_column_kwargs={"server_default": "0"}) started_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) finished_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) @@ -138,20 +142,26 @@ class TalentRuns(SQLModel, table=True): async def mark_rearmed( cls, session: AsyncSession, record_id, *, apify_run_id, apify_dataset_id, search_input: dict, found_so_far: int, + cost_usd: float | None = None, ): """Point the SAME run row at a broadened follow-up actor run. Status stays "running" so the frontend keeps polling and the 409 - active-run guard keeps holding; profiles_found accumulates across - the ladder's batches. + active-run guard keeps holding; profiles_found and cost_usd accumulate + across the ladder's batches (the caller folds the finished rung's + usageTotalUsd in BEFORE apify_run_id swaps — after the swap the old + run's cost is no longer reachable from this row). """ - return await cls._update(session, record_id, { + fields: dict = { "status": "running", "apify_run_id": apify_run_id, "apify_dataset_id": apify_dataset_id, "search_input": search_input, "profiles_found": found_so_far, - }) + } + if cost_usd is not None: + fields["cost_usd"] = cost_usd + return await cls._update(session, record_id, fields) @classmethod async def mark_status(cls, session: AsyncSession, record_id, status: str): @@ -161,21 +171,51 @@ class TalentRuns(SQLModel, table=True): return await cls._update(session, record_id, fields) @classmethod - async def mark_failed(cls, session: AsyncSession, record_id, error: str, *, status: str = "failed"): - return await cls._update(session, record_id, { + async def mark_failed( + cls, session: AsyncSession, record_id, error: str, *, + status: str = "failed", cost_usd: float | None = None, + ): + fields: dict = { "status": status, "apify_error": (error or "")[:2000], "finished_at": _now(), - }) + } + if cost_usd is not None: + fields["cost_usd"] = cost_usd + return await cls._update(session, record_id, fields) @classmethod - async def mark_succeeded(cls, session: AsyncSession, record_id, *, profiles_found: int): - return await cls._update(session, record_id, { + async def mark_succeeded( + cls, session: AsyncSession, record_id, *, + profiles_found: int, cost_usd: float | None = None, + ): + fields: dict = { "status": "succeeded", "profiles_found": profiles_found, "apify_error": None, "finished_at": _now(), - }) + } + if cost_usd is not None: + fields["cost_usd"] = cost_usd + return await cls._update(session, record_id, fields) + + @classmethod + async def cost_totals(cls, session: AsyncSession) -> tuple[float, int]: + """(total cost, total profiles) over succeeded runs with recorded cost. + + Rows from before 018_talent_run_cost.sql carry cost_usd = 0 and are + excluded so they cannot drag the $/profile average toward zero. + """ + statement = select( + func.coalesce(func.sum(cls.cost_usd), 0.0), + func.coalesce(func.sum(cls.profiles_found), 0), + ).where( + cls.status == "succeeded", + cls.is_deleted == False, # noqa: E712 + cls.cost_usd > 0, + ) + total_cost, total_profiles = (await session.execute(statement)).one() + return float(total_cost or 0.0), int(total_profiles or 0) class TalentProfiles(SQLModel, table=True): diff --git a/backend/talent/plugins.py b/backend/talent/plugins.py index a0409fc..1f49ab0 100644 --- a/backend/talent/plugins.py +++ b/backend/talent/plugins.py @@ -309,6 +309,55 @@ async def get_me() -> dict: return (response.json() or {}).get("data") or {} +async def get_limits() -> dict: + """GET /users/me/limits: monthly usage cycle, spend so far, and the cap. + Prepaid credits surface through maxMonthlyUsageUsd, so cap minus spend is + the account's remaining balance.""" + async with httpx.AsyncClient(timeout=APIFY_TIMEOUT) as client: + response = await client.get( + f"{APIFY_API_BASE}/users/me/limits", headers=_headers() + ) + _raise_for_response(response) + return (response.json() or {}).get("data") or {} + + +def run_cost_usd(remote: dict) -> float: + """What this Apify run actually charged, from the run object's + usageTotalUsd ("Represents what you actually pay" — covers pay-per-event + actors like HarvestAPI). 0.0 for anything absent or malformed.""" + try: + value = float((remote or {}).get("usageTotalUsd") or 0) + except (TypeError, ValueError): + return 0.0 + return value if value > 0 else 0.0 + + +def account_summary(limits_payload: dict) -> dict: + """Map the raw /users/me/limits payload to the fields the UI shows. + Every field is None when the payload lacks it; balance is clamped at 0 + because a mid-cycle limit reduction can leave spend above the cap.""" + payload = limits_payload or {} + current = payload.get("current") or {} + limits = payload.get("limits") or {} + cycle = payload.get("monthlyUsageCycle") or {} + + def _number(value) -> float | None: + try: + return float(value) + except (TypeError, ValueError): + return None + + spent = _number(current.get("monthlyUsageUsd")) + cap = _number(limits.get("maxMonthlyUsageUsd")) + balance = max(cap - spent, 0.0) if spent is not None and cap is not None else None + return { + "spent_this_cycle_usd": spent, + "monthly_limit_usd": cap, + "balance_usd": balance, + "cycle_ends_at": cycle.get("endAt"), + } + + def normalize_linkedin_url(url) -> str | None: """Canonical dedupe key: https, lowercase host/path, no query or trailing slash.""" text = str(url or "").strip() diff --git a/backend/talent/serializers.py b/backend/talent/serializers.py index fcda0e3..e80c0c3 100644 --- a/backend/talent/serializers.py +++ b/backend/talent/serializers.py @@ -10,6 +10,7 @@ def serialize_talent_run(row) -> dict: "search_input": row.search_input or {}, "max_results": row.max_results, "profiles_found": row.profiles_found, + "cost_usd": row.cost_usd or 0, "apify_run_id": row.apify_run_id, "apify_error": row.apify_error, "started_at": row.started_at.isoformat() if row.started_at else None, diff --git a/backend/talent/views.py b/backend/talent/views.py index b725edc..c17ed80 100644 --- a/backend/talent/views.py +++ b/backend/talent/views.py @@ -134,6 +134,11 @@ class Talent: raise HTTPException(status_code=502, detail=f"Apify status check failed: {exc}") status = plugins.local_status(remote.get("status")) + # Spend accumulates across the re-arm ladder: fold the current Apify + # run's charge into the row total only at a boundary (terminal or + # re-arm), because after a re-arm swaps apify_run_id the old run's + # cost is no longer reachable from this row. + total_cost = (run.cost_usd or 0.0) + plugins.run_cost_usd(remote) if status == "running": run = await TalentRuns.mark_status(self.session, run.id, "running") return serialize_talent_run(run) @@ -185,17 +190,20 @@ class Talent: apify_dataset_id=started.get("defaultDatasetId"), search_input=broadened, found_so_far=found_so_far, + cost_usd=total_cost, ) return serialize_talent_run(run) run = await TalentRuns.mark_succeeded( - self.session, run.id, profiles_found=found_so_far + self.session, run.id, profiles_found=found_so_far, cost_usd=total_cost ) return serialize_talent_run(run) # failed / timed_out / aborted message = remote.get("statusMessage") or f"Apify run {remote.get('status')}" - run = await TalentRuns.mark_failed(self.session, run.id, message, status=status) + run = await TalentRuns.mark_failed( + self.session, run.id, message, status=status, cost_usd=total_cost + ) return serialize_talent_run(run) async def fetch_runs(self, job_post_id): @@ -203,6 +211,25 @@ class Talent: rows, total = await TalentRuns.fetch_runs(self.session, job_post_id=job_post_id) return [serialize_talent_run(r) for r in rows], total + async def account(self): + """Apify account money for the Find Talent header: balance, spend this + cycle, and the observed $/profile. Balance/spend degrade to None when + Apify is unreachable (or the token is missing) so the screen still + renders; $/profile comes from our own recorded runs either way.""" + try: + summary = plugins.account_summary(await plugins.get_limits()) + except (httpx.HTTPError, plugins.ApifyError, RuntimeError): + summary = plugins.account_summary({}) + total_cost, total_profiles = await TalentRuns.cost_totals(self.session) + return { + **summary, + "total_cost_usd": total_cost, + "total_profiles_found": total_profiles, + "cost_per_profile_usd": ( + total_cost / total_profiles if total_profiles else None + ), + } + async def _outreach_actor_names(self, rows) -> dict: return await Users.names_by_ids( self.session, diff --git a/backend/tests/test_talent_plugins.py b/backend/tests/test_talent_plugins.py index e16c6d1..5e91e3d 100644 --- a/backend/tests/test_talent_plugins.py +++ b/backend/tests/test_talent_plugins.py @@ -490,3 +490,54 @@ def test_company_from_nested_experience_company_dict(): }) assert profile["current_title"] == "Engineer" assert profile["current_company"] == "Acme" + + +# ---------------------------------------------------------------- run cost + account + +def test_run_cost_reads_usage_total_usd(): + assert plugins.run_cost_usd({"usageTotalUsd": 0.2072}) == 0.2072 + assert plugins.run_cost_usd({"usageTotalUsd": "0.15"}) == 0.15 # API sends numbers, but be safe + + +def test_run_cost_is_zero_for_missing_or_malformed_values(): + assert plugins.run_cost_usd({}) == 0.0 + assert plugins.run_cost_usd(None) == 0.0 + assert plugins.run_cost_usd({"usageTotalUsd": None}) == 0.0 + assert plugins.run_cost_usd({"usageTotalUsd": "free"}) == 0.0 + assert plugins.run_cost_usd({"usageTotalUsd": -0.5}) == 0.0 # never subtract from a row total + + +def test_account_summary_maps_the_limits_payload(): + summary = plugins.account_summary({ + "monthlyUsageCycle": {"startAt": "2026-08-14T00:00:00Z", "endAt": "2026-09-14T00:00:00Z"}, + "limits": {"maxMonthlyUsageUsd": 5}, + "current": {"monthlyUsageUsd": 1.58}, + }) + assert summary == { + "spent_this_cycle_usd": 1.58, + "monthly_limit_usd": 5.0, + "balance_usd": 3.42, + "cycle_ends_at": "2026-09-14T00:00:00Z", + } + + +def test_account_summary_degrades_to_nones_on_empty_payload(): + # The /talent/account endpoint feeds this an empty dict when Apify is + # unreachable; every field must be None rather than a fake zero. + assert plugins.account_summary({}) == { + "spent_this_cycle_usd": None, + "monthly_limit_usd": None, + "balance_usd": None, + "cycle_ends_at": None, + } + assert plugins.account_summary(None)["balance_usd"] is None + + +def test_account_summary_clamps_balance_at_zero(): + # A mid-cycle limit reduction can leave spend above the cap. + summary = plugins.account_summary({ + "limits": {"maxMonthlyUsageUsd": 5}, + "current": {"monthlyUsageUsd": 7.5}, + }) + assert summary["balance_usd"] == 0.0 + assert summary["spent_this_cycle_usd"] == 7.5 diff --git a/frontend/src/api/talent.js b/frontend/src/api/talent.js index 074a72f..26cbf28 100644 --- a/frontend/src/api/talent.js +++ b/frontend/src/api/talent.js @@ -35,6 +35,15 @@ export function listRuns(jobPostId) { return request('/talent/runs/fetch', { params: { job_post_id: jobPostId } }) } +/** + * Apify account money: balance and spend this billing cycle (live from Apify, + * null when Apify is unreachable) plus the observed $/profile across all our + * recorded runs. Needs talent.view. + */ +export function getAccount() { + return request('/talent/account') +} + /** Sourced profiles for a job, most recently seen first. Needs talent.view. */ export function listProfiles({ jobId, search, top, skip } = {}) { return request('/talent/profiles/fetch', { @@ -80,6 +89,7 @@ export function toRunView(row) { status: row.status, maxResults: row.max_results ?? null, profilesFound: row.profiles_found ?? 0, + costUsd: row.cost_usd ?? null, error: row.apify_error ?? null, startedAt: row.started_at ? new Date(row.started_at) : null, finishedAt: row.finished_at ? new Date(row.finished_at) : null, @@ -88,6 +98,21 @@ export function toRunView(row) { } } +export function toAccountView(row) { + const r = row ?? {} + return { + balanceUsd: r.balance_usd ?? null, + spentUsd: r.spent_this_cycle_usd ?? null, + monthlyLimitUsd: r.monthly_limit_usd ?? null, + cycleEndsAt: r.cycle_ends_at ? new Date(r.cycle_ends_at) : null, + // null until at least one search has recorded its cost (runs from before + // cost tracking carry no spend data and are excluded from the average). + costPerProfileUsd: r.cost_per_profile_usd ?? null, + totalCostUsd: r.total_cost_usd ?? 0, + totalProfilesFound: r.total_profiles_found ?? 0, + } +} + export function toProfileView(row) { return { id: row.id, diff --git a/frontend/src/lib/queryKeys.js b/frontend/src/lib/queryKeys.js index e7eaa8a..c639220 100644 --- a/frontend/src/lib/queryKeys.js +++ b/frontend/src/lib/queryKeys.js @@ -84,6 +84,7 @@ export const qk = { }, talent: { all: () => ['talent'], + account: () => ['talent', 'account'], runs: (jobId) => ['talent', 'runs', jobId], run: (runId) => ['talent', 'run', runId], profiles: (p = {}) => ['talent', 'profiles', p], diff --git a/frontend/src/screens/Talent.jsx b/frontend/src/screens/Talent.jsx index 3c78622..944c649 100644 --- a/frontend/src/screens/Talent.jsx +++ b/frontend/src/screens/Talent.jsx @@ -41,6 +41,10 @@ const OUTREACH_TOAST = { sourced: 'Removed from shortlist', } +/* Apify money is in fractional dollars ($0.008/profile), so seed.js's + whole-dollar money() helper is the wrong tool here. */ +const fmtUsd = (n, digits = 2) => (n == null ? '—' : `$${n.toFixed(digits)}`) + async function fetchJobs() { const res = await candidatesApi.listJobs() const rows = Array.isArray(res?.data) ? res.data : [] @@ -451,6 +455,16 @@ export default function Talent() { const jobs = jobsQuery.data ?? [] const selectedJob = jobs.find((j) => j.id === jobId) + // Apify account money for the header chips. Keyed under ['talent', ...], so + // the settle-once effect's invalidation refreshes it after every search. + const accountQuery = useQuery({ + queryKey: qk.talent.account(), + queryFn: talentApi.getAccount, + }) + const account = accountQuery.data?.data + ? talentApi.toAccountView(accountQuery.data.data) + : null + const runsQuery = useQuery({ queryKey: qk.talent.runs(jobId), queryFn: () => talentApi.listRuns(jobId), @@ -542,6 +556,9 @@ export default function Talent() { setActiveRunId(run.id) } qc.invalidateQueries({ queryKey: qk.talent.runs(jobId) }) + // The settle effect refreshes the money chips when the run finishes; + // this catches the spend of the run that just started sooner. + qc.invalidateQueries({ queryKey: qk.talent.account() }) toast('Talent search started', 'success') }, onError: (err) => { @@ -575,15 +592,52 @@ export default function Talent() { LinkedIn Sourcing · Live} + actions={ + <> + LinkedIn Sourcing · Live + {account && ( + <> + + Balance {fmtUsd(account.balanceUsd)} + + + Spent {fmtUsd(account.spentUsd)} + + + {fmtUsd(account.costPerProfileUsd, 3)}/profile + + + )} + + } />
-
+
{locationChoice === CUSTOM_LOCATION && ( setCustomLocation(e.target.value)} @@ -636,11 +688,12 @@ export default function Talent() {

)} {jobId && statusRun && ( -

+

{badgeLabel} {statusRun.status === 'succeeded' && ( {statusRun.profilesFound} profile{statusRun.profilesFound === 1 ? '' : 's'} in the last run )} + {statusRun.costUsd > 0 && · cost {fmtUsd(statusRun.costUsd)}} {statusRun.error && {statusRun.error}} {statusRun.createdAt && · {fmtDate(statusRun.createdAt)}}

@@ -677,7 +730,7 @@ export default function Talent() { onChange={(t) => { setTab(t); setVisibleCount(10) }} tabs={tabs} /> -
+
This starts a paid Apify search of LinkedIn for people matching this job's title, technical requirements and experience level, {locationLabel} — - up to 25 profiles per run (roughly $0.20). Repeating the same search continues - deeper into the results, so each run surfaces new people; anyone already found - is refreshed, not duplicated. + up to 25 profiles per run ( + {account?.costPerProfileUsd != null + ? `≈ ${fmtUsd(account.costPerProfileUsd * 25)} at your average of ${fmtUsd(account.costPerProfileUsd, 3)}/profile` + : 'roughly $0.20'} + ). Repeating the same search continues deeper into the results, so each run + surfaces new people; anyone already found is refreshed, not duplicated.

)} diff --git a/frontend/src/styles/styles.css b/frontend/src/styles/styles.css index f91ebf1..405a4b2 100644 --- a/frontend/src/styles/styles.css +++ b/frontend/src/styles/styles.css @@ -1173,6 +1173,26 @@ canvas { width: 100%; max-width: 100%; display: block; } .cand-meta svg { width: 13px; height: 13px; } .cand-company { flex: 1; min-width: 0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; font-size: 12px; color: var(--text-3); } +/* Find Talent toolbar (Talent.jsx): the job picker takes the slack, the + location controls hold a readable fixed width. The widths live here, not + inline, so the ≤640 block can stack everything full-width. */ +.talent-controls { display: flex; align-items: center; gap: 8px; flex-wrap: wrap; } +.talent-controls .tc-job { flex: 1; min-width: 180px; } +.talent-controls .tc-loc { width: 220px; } +.talent-controls .tc-custom { width: 200px; } + +/* Find Talent header: Apify account money (balance / spent / $ per profile). + Same quiet-pill shape as .integration-status, but neutral — these are + numbers, not statuses. .page-head-actions already wraps them on phones. */ +.billing-chip { + display: inline-flex; align-items: center; gap: 5px; + padding: 6px 12px; border-radius: 20px; + font-size: var(--fs-sm); font-weight: 500; + background: var(--bg-sunken); border: 1px solid var(--border); + color: var(--text-2); white-space: nowrap; +} +.billing-chip strong { font-weight: 700; color: var(--text); font-variant-numeric: tabular-nums; } + /* Platform publish card */ .platform-card { display: flex; align-items: center; gap: 14px; padding: 16px; border: 1px solid var(--border); border-radius: 14px; transition: .15s; cursor: pointer; background: var(--bg-elev); } .platform-card:hover { border-color: var(--border-strong); box-shadow: var(--shadow-sm); } @@ -1507,6 +1527,16 @@ canvas { width: 100%; max-width: 100%; display: block; } .ai-dock { width: 100%; max-width: 100%; } .split-list { max-height: 320px; } + + /* Find Talent: stack the toolbar controls edge to edge. */ + .talent-controls .tc-job, .talent-controls .tc-loc, + .talent-controls .tc-custom, .talent-controls .btn { + flex: 1 1 100%; width: 100%; min-width: 0; + } + /* Candidate/profile card footer: a nowrap location label plus up to five + action buttons exceed a small phone's card width — let the row wrap + instead of pushing the last buttons past the card edge. */ + .cand-foot { flex-wrap: wrap; row-gap: 4px; } .card-head { padding: 14px 16px; } .card-body, .card-pad { padding: 16px; } .kpi { padding: 16px; }