Spaces:
Sleeping
Sleeping
"""Interview screen — build the profile by conversation instead of by files.
A user with no dossier folder (or a thin one) answers a guided set of
questions here and ends up with the same profile evidence the folder path
produces. The screen owns the conversation only: the question bank, the
answer validation, the follow-ups and the evidence build all live in
:mod:`app.services.interview`.
Answers are stored exactly as the user typed them: nothing here rewrites,
tidies or "cleans" what they wrote. The AI earns its place on the
conversation instead — checking an answer fits the question that was asked,
asking a follow-up when a needed detail is missing, phrasing the next
question, and acknowledging what was said.
On completion this writes exactly the session state a dossier read writes —
``dossier_chunks``, ``dossier_read``, ``dossier_read_error``,
``evidence_index``, ``canonical_profile``, ``evidence_index_meta`` — so every
screen downstream of Dossier works unchanged, plus ``profile_source`` so the
rest of the app can tell the two paths apart.
The interview is a second way through the Dossier step rather than a step of
its own: it renders under the Dossier rail row and the Dossier step number
(see ``STEP_PARENT`` in ``app/main.py``).
Nothing here is imported at provider level: the screen renders a clean
message and stays out of the way if the interview service is unavailable.
"""
from __future__ import annotations
import traceback
import copy
import re
from types import SimpleNamespace
from typing import Any
import streamlit as st
from app.config import clear_opportunity_state, get_settings
from app.services.evidence_index import build_evidence_index
from app.ui import theme
from app.utils.logging_utils import get_logger
_logger = get_logger("upwork_strategist.interview_screen")
try: # The interview service is a separate module; a missing or broken
# import must never take the rest of the app down with it.
from app.services import interview
except Exception: # noqa: BLE001 - the screen degrades to a clean message
interview = None # type: ignore[assignment]
# ---------------------------------------------------------------------------
# Session keys
# ---------------------------------------------------------------------------
#
# The first three are the contract keys other screens read. The last two are
# display-only state owned by this screen: the latest turn's reply, and a
# counter that forms part of the answer box's widget key so the box re-mounts
# empty after every send (the widget-reset idiom used elsewhere in this app).
_STATE_KEY = "interview_state"
_STARTED_KEY = "interview_started"
_SOURCE_KEY = "profile_source"
_LAST_TURN_KEY = "interview_last_turn"
_SEQ_KEY = "interview_turn_seq"
# The AI toggle's value, kept outside the widget. The intro card is not drawn
# while the interview runs, so Streamlit collects the toggle's own key, and a
# toggle seeded from ``value=`` alone springs back to its default the next
# time the card is drawn — silently overriding a choice the user made on
# purpose. This key is a plain value and survives that gap.
_AI_CHOICE_KEY = "interview_ai_choice"
_AI_TOGGLE_KEY = "interview_ai_toggle"
# Set when a Groq key is accepted on this screen. Streamlit keeps a widget's
# last value and ignores a new ``value=``, so without this the toggle would
# stay off — drawn disabled while there was no key — and the user who was
# just told "AI help is on" would start an interview without it.
_AI_FORCE_ON_KEY = "interview_ai_force_on"
# "Start over" throws away every answer given so far, so it asks first: this
# holds the "asked, not yet confirmed" step between the two clicks.
_RESTART_CONFIRM_KEY = "interview_restart_confirm"
# Below this many answered questions there is not enough to build a profile
# from, so "Finish early" stays disabled.
_MIN_ANSWERS_TO_FINISH = 3
# What a skip and a step backwards say. The skip line only tells the truth
# because "Previous" exists: without a way back there would be no "later".
_SKIP_ACK = "Skipped — you can come back to it with Previous."
_BACK_ACK = (
"Back to the previous question. Your earlier answer is in the box: edit it "
"and Send to replace it, or Skip to keep it."
)
# Which API the interview runs on this session: "" tries a Groq key first,
# "main" is set when the Groq check failed and the Setup API took over, and
# the note says so in words.
_ROUTE_PREF_KEY = "interview_route_pref"
_ROUTE_NOTE_KEY = "interview_route_note"
# How the user came in. "chooser" is "Interview me instead": the interview
# replaces any folder read earlier — unless the profile already carries this
# interview on top of that folder (see :func:`_joins_folder`). "gap_fill" is the
# thin-folder offer: the interview adds to the folder. Set by the dossier
# screen.
_ENTRY_KEY = "interview_entry"
# The longest answer the service stores. The box stops there with a visible
# counter, so nothing is ever cut without the user seeing it.
_MAX_ANSWER_CHARS = int(getattr(interview, "_MAX_ANSWER_CHARS", 1500) or 1500)
# Markdown punctuation. Model replies and typed answers are shown as text:
# a "$10 then $20" must not turn into maths, nor "*this*" into italics.
_MD_SPECIAL = re.compile(r"([\\`*_{}\[\]()#+\-.!|$<>~])")
def _plain(text: Any) -> str:
"""Escape Markdown so a dynamic message is shown exactly as written."""
return _MD_SPECIAL.sub(r"\\\1", str(text or ""))
# How many transcript entries stay on screen. The whole conversation is kept
# in state; only the tail is rendered so the page stays readable.
_TRANSCRIPT_TAIL = 8
_BUILD_FAILED_MESSAGE = (
"We couldn't turn those answers into a profile yet. Answer a few more "
"questions and try again."
)
# The interview provider is checked once, on the way in, and the verdict is
# cached under this key: the screen reruns on every turn, and a live probe
# per turn would spend the user's free quota on diagnostics.
_API_CHECK_KEY = "interview_api_check"
# Where the interview publishes its own strength score. ``dossier_validation``
# describes a folder of files and is never written from here.
_STRENGTH_KEY = "interview_strength"
# Status strings from ``app.services.api_gate.ApiGateError``. Spelled out
# rather than imported so this module keeps importing with no gate present.
_STATUS_OK = "API OK"
_STATUS_INVALID_KEY = "INVALID API KEY"
_STATUS_QUOTA = "API QUOTA EXCEEDED"
_STATUS_NO_KEY = "NO API ADDED"
_STATUS_NO_MODEL = "MODEL NOT CONFIGURED"
_STATUS_CONNECTION = "API CONNECTION FAILED"
# What the user is told when the check comes back as anything but OK. They
# all end the same way, because none of them stops the interview.
_AI_OFF_REASONS: dict[str, str] = {
"MODEL NOT ACCEPTED": "The provider rejected the interview model. Choose another model in Setup. The interview can continue without AI help.",
"API ACCESS DENIED": "Your key cannot access the interview model. Check permissions or choose another model in Setup. The interview can continue without AI help.",
_STATUS_INVALID_KEY: (
"The API key the interview uses was rejected, so it asks its questions as "
"written, with only basic answer checks and no follow-ups."
),
_STATUS_QUOTA: (
"The interview's API quota is used up for now, so it asks its "
"questions as written, with only basic answer checks and no follow-ups."
),
_STATUS_NO_KEY: (
"No API key is available for the interview, so it asks its questions "
"as written, with only basic answer checks and no follow-ups."
),
_STATUS_NO_MODEL: (
"No interview model is configured, so the interview asks its "
"questions as written, with only basic answer checks and no follow-ups."
),
_STATUS_CONNECTION: (
"The interview service couldn't be reached, so it asks its questions "
"as written, with only basic answer checks and no follow-ups."
),
}
_AI_OFF_FALLBACK = (
"No AI service is set up for the interview, so it asks its questions as "
"written, with only basic answer checks and no follow-ups."
)
# Said alongside every one of those: what the AI is off changes, and what it
# never changed.
_ANSWERS_VERBATIM = "Your answers are saved exactly as you type them either way."
# ---------------------------------------------------------------------------
# Availability
# ---------------------------------------------------------------------------
def is_available() -> bool:
"""True when the interview path can be offered to the user.
The dossier screen asks before it advertises the interview, so a missing
service simply leaves the existing folder flow exactly as it was.
"""
return interview is not None
def show_intro_on_entry() -> None:
"""Send the next visit to the interview back to its intro card.
``interview_started`` is what hides that card, and nothing else clears
it, so a user who left a part-finished interview and came back would
land straight on the live question with the AI toggle and "Start over"
out of reach. No answer is touched: with a state in hand the card's
button reads "Resume the interview".
"""
st.session_state[_STARTED_KEY] = False
# A "Start over" that was asked but never confirmed on an earlier visit
# must not be sitting there with its confirm button live.
st.session_state[_RESTART_CONFIRM_KEY] = False
def _run_api_check(settings: Any, prefer: str = "") -> Any:
"""Probe the interview provider and return the gate's verdict.
Imported lazily so this module stays importable with no provider set
up. The probe is live because that is the only way to tell a rejected
key from an exhausted free tier, and the two need different words.
"""
try:
from app.services.api_gate import run_interview_capability_test
except Exception: # noqa: BLE001 - older gate without the interview check
return None
kwargs: dict = {"settings": settings, "live": True}
if prefer:
kwargs["prefer"] = prefer
try:
with st.spinner("Checking the interview service…"):
return run_interview_capability_test(**kwargs)
except Exception: # noqa: BLE001 - never let a probe break the screen
return None
def _status_of(verdict: Any) -> str:
return str(getattr(verdict, "status", "") or "")
def _routes(settings: Any) -> tuple[Any, Any]:
"""The default interview route, and the Setup API route without Groq."""
try:
from app.services import llm_client # lazy: importable without it
return (
llm_client.interview_route(settings),
llm_client.interview_route(settings, prefer="main"),
)
except Exception: # noqa: BLE001 - a missing router means no routes
return None, None
_FAILURE_WORDS = {
_STATUS_INVALID_KEY: "the key was rejected",
_STATUS_QUOTA: "its free quota is used up",
_STATUS_NO_MODEL: "its model isn't available",
_STATUS_CONNECTION: "it couldn't be reached",
}
def _probe(settings: Any) -> Any:
"""Check the interview's API, falling back from Groq to the Setup API.
A Groq key wins when one is set, but when its check fails and the Setup
API works, the interview runs on the Setup API instead of losing its AI
— and the note says why. The choice is kept for the session.
"""
st.session_state[_ROUTE_PREF_KEY] = ""
st.session_state.pop(_ROUTE_NOTE_KEY, None)
verdict = _run_api_check(settings)
status = _status_of(verdict)
if status == _STATUS_OK:
return verdict
default, main = _routes(settings)
if default and main and default[0] == "groq" and main[0] != "groq":
fallback = _run_api_check(settings, prefer="main")
if _status_of(fallback) == _STATUS_OK:
st.session_state[_ROUTE_PREF_KEY] = "main"
name = _PROVIDER_NAMES.get(str(main[0]), str(main[0]).title())
why = _FAILURE_WORDS.get(status, "it couldn't be used")
st.session_state[_ROUTE_NOTE_KEY] = (
f"Your Groq key couldn't be used ({why}), so the interview "
f"uses your {name} API from Setup instead."
)
return fallback
return verdict
def _checked() -> bool:
"""True once this session has a verdict — even an empty one — in hand.
What separates "the AI is unavailable" from "nobody has looked yet". The
intro card is drawn before anyone has looked, and must not start a
network call just because the user opened it.
"""
return _API_CHECK_KEY in st.session_state
def _api_check(settings: Any) -> Any:
"""This session's cached verdict, running the probe on first use.
The key is written even when the probe returns nothing, so a provider
that is down is asked exactly once per session and never again on the
reruns that carry the conversation.
"""
if _API_CHECK_KEY not in st.session_state:
st.session_state[_API_CHECK_KEY] = _probe(settings)
return st.session_state.get(_API_CHECK_KEY)
def _ai_status(settings: Any) -> str:
return str(getattr(_api_check(settings), "status", "") or "")
def _ai_ready(settings: Any) -> bool:
"""True when the interview's AI turns can actually run."""
return _ai_status(settings) == _STATUS_OK
def _cached_status() -> str:
"""The verdict already in hand, or "" — never starts a probe."""
return str(getattr(st.session_state.get(_API_CHECK_KEY), "status", "") or "")
def _cached_ai_ready() -> bool:
"""True when a verdict is in hand and it says the AI turns can run."""
return _cached_status() == _STATUS_OK
def _ai_off_reason() -> str:
"""Why the AI turns are off, in the user's words — "" when they are on.
Reads the verdict already in hand: it is only ever asked for after
something has looked.
"""
status = _cached_status()
if status == _STATUS_OK:
return ""
return f"{_AI_OFF_REASONS.get(status, _AI_OFF_FALLBACK)} {_ANSWERS_VERBATIM}"
# ---------------------------------------------------------------------------
# State
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Groq key, entered on this screen
# ---------------------------------------------------------------------------
# A one-shot (kind, text) message shown after a key is saved, because the
# save reruns the page and a message drawn before the rerun would be lost.
_KEY_FLASH_KEY = "interview_key_flash"
def _interview_key_env() -> str:
"""Env name holding the interview key, e.g. ``GROQ_API_KEY``."""
provider = str(getattr(interview, "INTERVIEW_PROVIDER", "groq") or "groq")
return f"{provider.strip().upper() or 'GROQ'}_API_KEY"
_PROVIDER_NAMES = {
"anthropic": "Claude",
"openai": "OpenAI",
"groq": "Groq",
"gemini": "Gemini",
}
def _route_name(settings: Any) -> tuple[str, bool]:
"""The interview's provider as the user knows it, and whether it is Setup's.
Returns ``("", False)`` when no provider is available. The second value
is True when the interview runs on the same API configured on Setup,
False when a separately added Groq key has taken over.
"""
pref = str(st.session_state.get(_ROUTE_PREF_KEY, "") or "")
try:
from app.services import llm_client # lazy: importable without it
route = (
llm_client.interview_route(settings, prefer=pref)
if pref
else llm_client.interview_route(settings)
)
except Exception: # noqa: BLE001 - a missing router means no provider
route = None
if not route:
return "", False
provider = str(route[0] or "")
main = str(getattr(settings, "llm_provider", "") or "").strip().lower()
return _PROVIDER_NAMES.get(provider, provider.title()), provider == main
def _has_interview_key(settings: Any) -> bool:
try:
return bool(interview.interview_available(settings)) # type: ignore[union-attr]
except Exception: # noqa: BLE001 - a missing helper means no key
return False
def _key_is_main_key(settings: Any) -> bool:
"""True when the interview's key slot is the one the main API runs on.
With Groq picked on Setup, a key typed on this screen would replace the
key every other step uses, and a mistyped one would erase it. Setup
hides its own interview-key field in this case for the same reason.
"""
provider = str(getattr(interview, "INTERVIEW_PROVIDER", "groq") or "groq")
main = str(getattr(settings, "llm_provider", "") or "")
return main.strip().lower() == provider.strip().lower()
def _needs_key_form(settings: Any, state: Any) -> bool:
"""True when the AI can't run for want of a working key.
Not shown when the AI is off for a reason a key can't fix — the user
turned it off with a working key in hand, or this interview has used up
its own call allowance.
"""
if _cached_ai_ready():
return False
if state is not None:
reason = str(getattr(state, "ai_off_reason", "") or "")
if reason == getattr(interview, "AI_OFF_BUDGET", "budget_spent"):
return False
if _checked():
return _cached_status() in (
_STATUS_NO_KEY,
_STATUS_INVALID_KEY,
_STATUS_QUOTA,
_STATUS_NO_MODEL,
_STATUS_CONNECTION,
"",
)
return not _has_interview_key(settings)
def _remember_config() -> None:
"""Remember the session's keys on this device, when storage is set up."""
try:
from app.ui import setup_screen # lazy: avoids an import cycle
setup_screen.remember_config()
except Exception: # noqa: BLE001 - remembering is a nicety, never a blocker
pass
def _save_interview_key(raw: str, state: Any) -> None:
"""Apply a typed Groq key, test it live, and act on the verdict. Caller reruns.
The key goes into this session's memory the same way the Setup screen
puts it there, but the main API check is left alone: this key only ever
gates the interview's AI turns, so saving it must not send the user
back to Setup. It is remembered on the device only once Groq has
answered for it: accepted, or out of quota.
"""
from app import config # lazy: keeps this module importable anywhere
key = (raw or "").strip()
if not key:
st.session_state[_KEY_FLASH_KEY] = (
"warning",
"Paste your Groq key into the box first. It starts with gsk_.",
)
return
env = _interview_key_env()
# What the slot held before, so a rejected key can be undone rather than
# leaving the session without a key it had.
previous = str(config.export_session_overrides().get(env, "") or "")
config.apply_session_overrides({env: key})
st.session_state.pop(_API_CHECK_KEY, None)
settings = get_settings()
groq_verdict = _run_api_check(settings)
groq_status = _status_of(groq_verdict)
if groq_status == _STATUS_OK:
_remember_config()
st.session_state[_API_CHECK_KEY] = groq_verdict
st.session_state[_ROUTE_PREF_KEY] = ""
st.session_state.pop(_ROUTE_NOTE_KEY, None)
_ai_back_on(state, route_pref="")
st.session_state[_KEY_FLASH_KEY] = (
"success",
"Groq key accepted — AI help is on. It now checks each answer and "
"asks a follow-up when something is missing.",
)
return
if groq_status == _STATUS_INVALID_KEY:
# A rejected key is not kept: leaving it would make the app believe a
# key is set up when none works. The slot goes back to what it held,
# so a mistyped paste never costs a key the session already had —
# unless that was the key just rejected.
config.apply_session_overrides({env: previous if previous != key else ""})
settings = get_settings()
elif groq_status == _STATUS_QUOTA:
# Rate-limited: Groq answered for the key, so keep and remember it.
_remember_config()
# Unreachable or unanswered, the key may be fine, so it is kept — for
# this session only, because nothing has confirmed it yet.
# The Setup API can still carry the interview.
verdict = _probe(settings)
st.session_state[_API_CHECK_KEY] = verdict
why = _FAILURE_WORDS.get(groq_status, "it couldn't be used")
if _status_of(verdict) == _STATUS_OK:
name, _ = _route_name(settings)
_ai_back_on(state, route_pref=str(st.session_state.get(_ROUTE_PREF_KEY, "") or ""))
st.session_state[_KEY_FLASH_KEY] = (
"warning",
f"That Groq key couldn't be used ({why}), so the interview keeps "
f"using your {name or 'Setup'} API.",
)
return
# Nothing works: keep the failed verdict so the box stays for another try.
st.session_state[_API_CHECK_KEY] = groq_verdict or verdict
if groq_status == _STATUS_INVALID_KEY:
text = "Groq didn't accept that key. Check you copied the whole key and try again."
kind = "error"
elif groq_status == _STATUS_QUOTA:
text = (
"That key works, but its free Groq quota is used up for today. Try "
"another key, or carry on without AI help for now."
)
kind = "warning"
else:
text = (
"Couldn't reach Groq just now, so the key isn't confirmed yet. Try "
"again in a moment — the interview carries on meanwhile."
)
kind = "warning"
st.session_state[_KEY_FLASH_KEY] = (kind, text)
def _ai_back_on(state: Any, *, route_pref: str) -> None:
"""Switch the AI turns back on after a working API was found."""
st.session_state[_AI_CHOICE_KEY] = True
st.session_state[_AI_FORCE_ON_KEY] = True
if state is not None:
state.ai_mode = "on"
state.ai_off_reason = ""
state.ai_off_status = ""
state.soft_failures = 0
state.route_pref = route_pref
# The explanation for the AI being off no longer applies.
last = dict(st.session_state.get(_LAST_TURN_KEY) or {})
last["ack"] = ""
last["note"] = ""
st.session_state[_LAST_TURN_KEY] = last
def _show_key_flash() -> None:
flash = st.session_state.pop(_KEY_FLASH_KEY, None)
if isinstance(flash, tuple) and len(flash) == 2:
kind, text = flash
{"success": st.success, "error": st.error}.get(kind, st.warning)(text)
def _key_form(settings: Any, state: Any, *, where: str) -> None:
"""A box and a button that take a Groq key and turn the AI on.
Drawn wherever the interview can end up without its AI: the intro card
and the live question. A form, so the key is sent with one click on the
button or one press of Enter, and the box empties once it is used.
"""
_show_key_flash()
if not _needs_key_form(settings, state):
return
if _key_is_main_key(settings):
# The box would write over the main key, so the key is changed where
# it lives instead.
st.caption(
"The interview uses your Groq key from Setup. To use a different "
"key, change it on Setup."
)
return
try:
from app.services import key_store
remembered = bool(key_store.is_configured())
except Exception: # noqa: BLE001
remembered = False
with st.form(key=f"iv_groq_form_{where}", clear_on_submit=True, border=False):
st.markdown("**Turn on AI help with a free Groq key**")
col_key, col_btn = st.columns([3, 1], vertical_alignment="bottom")
typed = col_key.text_input(
"Groq API key",
type="password",
placeholder="gsk_… paste your free Groq key here",
key=f"iv_groq_key_{where}",
label_visibility="collapsed",
)
submitted = col_btn.form_submit_button(
"Turn on AI", type="primary", use_container_width=True
)
st.caption(
"Get one free at console.groq.com/keys. It's used only for this "
"interview"
+ (
" and is remembered, encrypted, on this device."
if remembered
else " and kept for this session."
)
)
# Streamlit's "Press Enter to submit form" hint overlaps the masked key
# and the show/hide button at phone width; the button says the same.
theme.html(
f"<style>.st-key-iv_groq_form_{where} "
'[data-testid="InputInstructions"]{display:none}</style>'
)
if submitted:
_save_interview_key(typed, state)
st.rerun()
def _new_state(ai_mode: str) -> Any:
"""A fresh interview state, set to the AI mode the user chose."""
state = interview.InterviewState() # type: ignore[union-attr]
state.ai_mode = ai_mode
return state
def _reset_turn_display(ack: str = "") -> None:
st.session_state[_LAST_TURN_KEY] = {
"kind": "next",
"message": "",
"ack": ack,
"note": "",
}
def _resume_turn_display(state: Any, ack: str = "") -> None:
"""The turn display for an interview picked up again from the intro card.
A follow-up asked before the user left is still waiting for its reply:
the service takes the next answer as that reply, unchecked, and adds it
to the answer before it. So the follow-up is shown again — drawn as the
bare question, it would be answered a second time and never get its
reply. A follow-up that can no longer be found is dropped instead, and
the question opens with its stored answer in the box to be replaced.
"""
_reset_turn_display(ack)
if not bool(getattr(state, "pending_follow", False)):
return
# The service's own lookup: the very follow-up the reply is sent with.
finder = getattr(interview, "_last_follow_up", None)
try:
follow_up = str(finder(state) or "") if callable(finder) else ""
except Exception: # noqa: BLE001 - an odd transcript asks the question again
follow_up = ""
if follow_up:
st.session_state[_LAST_TURN_KEY].update(kind="follow_up", message=follow_up)
return
state.pending_follow = False
index = int(getattr(state, "index", 0) or 0)
if str((getattr(state, "answers", {}) or {}).get(index, "") or "").strip():
state.editing = index
def _start(ai_mode: str, *, ack: str = "") -> None:
"""Begin (or restart) the interview. Caller reruns.
``ack`` is shown above the first question: it is how a user who asked for
the AI and could not have it finds that out, rather than wondering why no
follow-up ever comes.
"""
st.session_state[_STATE_KEY] = _new_state(ai_mode)
st.session_state[_STARTED_KEY] = True
st.session_state[_RESTART_CONFIRM_KEY] = False
st.session_state[_SEQ_KEY] = int(st.session_state.get(_SEQ_KEY, 0)) + 1
_reset_turn_display(ack)
def _questions() -> list:
return list(getattr(interview, "QUESTIONS", None) or [])
def _current_question(state: Any) -> tuple[Any, int]:
"""The question the interview is on, plus its position in the bank."""
questions = _questions()
index = int(getattr(state, "index", 0) or 0)
if index < 0 or index >= len(questions):
return None, index
return questions[index], index
def _answer_count(state: Any) -> int:
"""How many questions count as answered.
Delegated to the service so the Finish gate, the caption and the
profile build agree. A declined question ("none", "skip", "n/a") is
stored but is not an answer, and three declines must not unlock a
build that comes out empty.
"""
try:
return int(interview.progress(state)["answered"]) # type: ignore[union-attr]
except Exception: # noqa: BLE001 - an odd state must not break the header
return 0
def _is_done(state: Any) -> bool:
if bool(getattr(state, "done", False)):
return True
questions = _questions()
return bool(questions) and int(getattr(state, "index", 0) or 0) >= len(questions)
# ---------------------------------------------------------------------------
# Turns
# ---------------------------------------------------------------------------
def _payload(result: Any) -> dict:
"""One turn's result, read defensively into the display dict."""
return {
"kind": str(getattr(result, "kind", "next") or "next"),
"message": str(getattr(result, "message", "") or ""),
"ack": str(getattr(result, "ack", "") or ""),
"note": str(getattr(result, "note", "") or ""),
}
def _land(payload: dict) -> None:
"""Show one turn's result and rerun.
The rerun is what makes a single click land: the top bar, the rail and
the profile panel are all drawn before this screen runs, so they only
show the new turn on the next pass.
"""
st.session_state[_LAST_TURN_KEY] = payload
# The counter is what re-mounts the answer box empty, so it only moves
# when the turn actually consumed the answer. A retry is the service
# handing the same answer back to be improved: wiping it there would
# delete the user's words at the exact moment they are asked to edit
# them.
if payload["kind"] != "retry":
st.session_state[_SEQ_KEY] = int(st.session_state.get(_SEQ_KEY, 0)) + 1
st.rerun()
_OFF_STATUS_VERDICTS = {
"invalid_key": _STATUS_INVALID_KEY,
"quota": _STATUS_QUOTA,
"no_api": _STATUS_NO_KEY,
"model_missing": "MODEL NOT ACCEPTED",
"access_denied": "API ACCESS DENIED",
"unsupported_provider": _STATUS_NO_MODEL,
}
def _after_ai_lost(state: Any, settings: Any) -> None:
"""A turn found the provider gone: say why, and try the fallback route.
The cached verdict said "API OK", which would hide both the reason and
the key box, so it is replaced with what the turn actually saw. When a
Groq key was what failed and the Setup API works, the AI comes straight
back on through the Setup API.
"""
off_reason = str(getattr(state, "ai_off_reason", "") or "")
if getattr(state, "ai_mode", "on") != "off" or off_reason != getattr(
interview, "AI_OFF_PROVIDER", "provider_unavailable"
):
return
status = _OFF_STATUS_VERDICTS.get(
str(getattr(state, "ai_off_status", "") or ""), _STATUS_CONNECTION
)
st.session_state[_API_CHECK_KEY] = SimpleNamespace(status=status)
if not getattr(state, "route_pref", ""):
default, main = _routes(settings)
if default and main and default[0] == "groq" and main[0] != "groq":
verdict = _probe(settings)
if _status_of(verdict) == _STATUS_OK:
st.session_state[_API_CHECK_KEY] = verdict
_ai_back_on(state, route_pref=str(st.session_state.get(_ROUTE_PREF_KEY, "") or ""))
def _submit(state: Any, text: str, *, from_chip: bool) -> None:
"""Send one answer to the interview service, then rerun.
The turn runs on a copy of the state, which takes the session's place
only once the turn is complete. A second click on Send while the model
is still reading starts a new run at once, and Streamlit stops this one
at its next touch of the session, part-way through the turn. Changed in
place, the state the new run reads would already hold this run's half
of the turn: the answer would be taken twice, and a follow-up it replies
to forgotten. On a copy, a stopped run — or a failed turn — leaves the
state exactly as it found it, and the swap is itself a session write,
which is where a stopped run is cut off.
"""
settings = get_settings()
turn = copy.deepcopy(state)
turn.route_pref = str(st.session_state.get(_ROUTE_PREF_KEY, "") or "")
try:
with st.spinner("Reading your answer…"):
result = interview.submit_answer( # type: ignore[union-attr]
turn, text, settings=settings, from_chip=from_chip
)
except Exception: # noqa: BLE001 - a failed turn must not crash the screen
_logger.error("Interview turn failed: %s", traceback.format_exc())
payload = {
"kind": "retry",
"message": "Something went wrong with that answer. Send it again.",
"ack": "",
"note": "",
}
else:
payload = _payload(result)
_after_ai_lost(turn, settings)
# A fallback may have switched the AI back on after the service wrote
# its "working without AI" note; the note would now be wrong.
if getattr(turn, "ai_mode", "off") == "on":
payload["note"] = ""
st.session_state[_STATE_KEY] = turn
_land(payload)
def _skip(state: Any) -> None:
"""Move past the current question without answering it, then rerun.
The move belongs to the service: :func:`interview.skip_question` files
nothing under the question's key, validates nothing and follows nothing
up, and records the skip in the transcript so the conversation still
reads straight. Its result then goes through the same display path a
sent answer takes, so a skip that lands on the last question ends the
interview exactly the way a final answer does.
"""
try:
result = interview.skip_question(state) # type: ignore[union-attr]
except Exception: # noqa: BLE001 - a failed skip must not crash the screen
_logger.error("Interview skip failed: %s", traceback.format_exc())
payload = {
"kind": "retry",
"message": "Something went wrong skipping that one. Try again.",
"ack": "",
"note": "",
}
else:
payload = _payload(result)
payload["ack"] = _SKIP_ACK
_land(payload)
def _previous(state: Any) -> None:
"""Step back to the question before this one, then rerun.
Any pending follow-up is dropped because it belonged to the question
being left. The earlier answer stays in state and comes back in the box
to be edited; sending replaces it, and Skip keeps it as it was.
"""
index = int(getattr(state, "index", 0) or 0)
if index <= 0:
return
back = getattr(interview, "previous_question", None)
if callable(back):
back(state)
else: # an older service: move the cursor only
state.pending_follow = False
state.done = False
state.index = min(index, len(_questions())) - 1
st.session_state[_SEQ_KEY] = int(st.session_state.get(_SEQ_KEY, 0)) + 1
_reset_turn_display(_BACK_ACK)
st.rerun()
# ---------------------------------------------------------------------------
# Completion
# ---------------------------------------------------------------------------
def _strength(state: Any) -> dict:
try:
return dict(interview.profile_strength(state) or {}) # type: ignore[union-attr]
except Exception: # noqa: BLE001 - the panel degrades to zero, never errors
return {}
def _strength_tone(score: int) -> str:
"""The strength label's tone — same bands as the dossier screen's."""
if score >= 80:
return "ok"
if score >= 60:
return "t2"
if score >= 40:
return "warn"
return "stop"
def _strength_band(score: int) -> str:
if score >= 70:
return "ok"
if score >= 40:
return "warn"
return "stop"
def _record_strength(state: Any) -> None:
"""Publish the interview's own strength score, under its own key.
``dossier_validation`` is a reading of a folder of files: its score,
its breakdown and its file list describe that folder and nothing else.
Raising its score from here would leave an object claiming a strength
those files never earned, with a breakdown that no longer adds up to
it, so the folder's result is left exactly as the folder left it.
"""
score = int(_strength(state).get("score", 0) or 0)
st.session_state[_STRENGTH_KEY] = max(0, min(100, score))
def _is_interview_chunk(chunk: Any) -> bool:
"""True when this chunk is one the interview itself wrote."""
name = str(
getattr(interview, "PROFILE_FILE_NAME", "interview_profile.json")
).strip().lower()
for attr in ("file_name", "file_path"):
value = (
chunk.get(attr) if isinstance(chunk, dict) else getattr(chunk, attr, "")
)
if str(value or "").strip().lower() == name:
return True
return False
def _interview_meta(state: Any) -> dict:
"""The interview's own keys on ``evidence_index_meta``.
``interview.build_evidence`` writes these on the pure-interview path.
The gap-fill path re-indexes the folder's chunks and the interview chunk
together through the folder's own builder, which knows nothing about an
interview, so the same keys are merged in by hand there — otherwise the
two paths hand downstream screens meta of two different shapes.
"""
return {
"source": "interview",
"error_message": (
"Profile built from your files plus the guided interview; "
"evidence extracted with the local JSON scanner."
),
"interview_strength": int(_strength(state).get("score", 0) or 0),
"interview_answers": _answer_count(state),
# Only this path carries the folder's files. Analysis reads it to
# decide whether the folder's own score still speaks for the profile.
"with_folder": True,
}
def _joins_folder() -> bool:
"""True when the interview is added to a folder's files, not put in their place.
"Interview me instead" replaces a folder read earlier; only the
thin-folder offer adds the interview to the folder. Sessions from before
the entry was recorded keep the old behaviour.
"""
ss = st.session_state
entry = str(ss.get(_ENTRY_KEY, "") or "")
# The profile in use has the last word over the button, though. Once a
# gap-filling interview is built into the folder, Dossier stops offering
# the gap fill, so "Interview me instead" is the only way back to carry
# on with it — and carrying on is not asking to throw the folder away.
meta_now = ss.get("evidence_index_meta")
merged = (
ss.get(_SOURCE_KEY) == "interview"
and isinstance(meta_now, dict)
and bool(meta_now.get("with_folder"))
)
return entry != "chooser" or merged
def adds_to_folder() -> bool:
"""True when the profile in use is this interview, kept with the folder's files.
The dossier screen asks before it reads the folder again. That read
builds the profile from the files alone, so unless the interview is
added back (:func:`rebuild_profile`), a gap-filling interview's answers
drop out of the profile — and out of every proposal — without a word.
"""
ss = st.session_state
return (
ss.get(_SOURCE_KEY) == "interview"
and ss.get(_STATE_KEY) is not None
and _joins_folder()
)
def rebuild_profile(folder_chunks: Any = None, *, state: Any = None) -> bool:
"""Build the profile from the interview's answers, on top of a folder's chunks.
With ``folder_chunks``, the interview chunk is added to them and the
whole set is re-indexed, so a gap-filling interview strengthens the
folder rather than replacing it; without, the profile is the answers
alone. Either way it writes exactly the session state a dossier read
writes, plus ``profile_source`` and the interview's strength, and drops
what was built on the previous profile.
:func:`_finish` builds through here, and so does the dossier screen when
it reads a gap-filled folder again or loses its files, so the two
routes can never build the same profile two different ways. Returns
False, with the session untouched, when no profile comes out of it.
"""
ss = st.session_state
if state is None:
state = ss.get(_STATE_KEY)
if interview is None or state is None:
return False
# Only the folder's own chunks are something to gap-fill. A chunk this
# interview wrote on an earlier pass is dropped rather than built on:
# keeping it would append a second copy of the same chunk_id every time
# the profile is built and double the proof points. Dropping it is also
# what makes the rebuild idempotent — building twice gives one index.
base_chunks = [
chunk for chunk in (folder_chunks or []) if not _is_interview_chunk(chunk)
]
try:
chunks = list(interview.to_chunks(state)) # type: ignore[union-attr]
if base_chunks:
chunks = base_chunks + chunks
proofs, profile, meta = build_evidence_index(chunks)
meta = dict(meta or {})
meta.update(_interview_meta(state))
else:
proofs, profile, meta = (
interview.build_evidence(state) # type: ignore[union-attr]
)
except Exception: # noqa: BLE001 - the caller decides what to say
return False
if not proofs:
return False
ss["dossier_chunks"] = chunks
ss["dossier_read"] = True
ss["dossier_read_error"] = False
ss["evidence_index"] = proofs
ss["canonical_profile"] = profile
ss["evidence_index_meta"] = meta
ss[_SOURCE_KEY] = "interview"
# The job fields were read off a screenshot against the previous profile,
# and ``dossier_screen._reset_downstream`` drops them whenever the folder
# route rebuilds. This route rebuilds the same evidence, so it drops the
# same key: a profile built either way must leave the same state behind.
ss["extracted_job_fields"] = None
_record_strength(state)
# New evidence means the verdict and the proposal built on the old
# evidence are stale, and a stale proposal left on screen reads as the
# current one. ``dossier_screen._reset_downstream`` clears them for
# exactly this reason whenever the folder route rebuilds; the interview
# route rebuilds the same keys, so it clears them too.
clear_opportunity_state(ss)
return True
def _finish(state: Any) -> None:
"""Build the evidence index from the interview and go to Job Screenshot.
When a dossier was already read, the interview chunk is added to it and
the whole set is re-indexed, so a gap-filling interview strengthens the
folder rather than replacing it (see :func:`rebuild_profile`).
"""
ss = st.session_state
# Only the folder's own chunks are something to gap-fill: a chunk this
# interview wrote on an earlier pass is not a folder.
base_chunks = [
chunk
for chunk in (ss.get("dossier_chunks") or [])
if not _is_interview_chunk(chunk)
]
gap_fill = bool(base_chunks and ss.get("dossier_read")) and _joins_folder()
with st.spinner("Building your profile…"):
built = rebuild_profile(base_chunks if gap_fill else None, state=state)
if not built:
# An empty index would leave the next step locked, so stop here with a
# message instead of advancing into a dead end.
st.error(_BUILD_FAILED_MESSAGE)
return
ss["current_step"] = "screenshot"
st.rerun()
# ---------------------------------------------------------------------------
# Read-only views
# ---------------------------------------------------------------------------
def _stringify(value: Any, *, limit: int = 120) -> str:
"""One short line for a profile value of any shape."""
if value is None:
return ""
if isinstance(value, (list, tuple, set)):
text = ", ".join(str(v).strip() for v in value if str(v).strip())
elif isinstance(value, dict):
text = ", ".join(f"{k}: {v}" for k, v in value.items() if v)
else:
text = str(value).strip()
text = " ".join(text.split())
if len(text) > limit:
text = text[: limit - 1].rstrip() + "…"
return text
def _profile_pairs(state: Any, *, limit: int = 12) -> list[tuple[str, str]]:
try:
profile = interview.build_profile_dict(state) or {} # type: ignore[union-attr]
except Exception: # noqa: BLE001 - the panel is a preview, never a blocker
return []
pairs: list[tuple[str, str]] = []
for key, value in profile.items():
text = _stringify(value)
if not text:
continue
pairs.append((str(key).replace("_", " ").capitalize(), text))
if len(pairs) >= limit:
break
return pairs
# The transcript is written by the service; read it defensively so a change
# of wording in its entries can never break the screen.
_ROLE_KEYS = ("role", "speaker", "who", "from")
_TEXT_KEYS = ("text", "content", "message", "answer", "body")
_USER_ROLES = {"user", "you", "candidate", "freelancer", "human"}
def _transcript_items(state: Any) -> list[dict]:
entries = list(getattr(state, "transcript", None) or [])
items: list[dict] = []
for entry in entries[-_TRANSCRIPT_TAIL:]:
if not isinstance(entry, dict):
continue
role = ""
for key in _ROLE_KEYS:
if entry.get(key):
role = str(entry[key])
break
text = ""
for key in _TEXT_KEYS:
if entry.get(key):
text = str(entry[key])
break
if not text.strip():
continue
who = "You" if role.strip().lower() in _USER_ROLES else "Interviewer"
items.append({"title": who, "body": text.strip()})
return items
def _render_transcript(state: Any) -> None:
"""The conversation so far.
Deliberately not inside a bordered panel: a keyed panel replays its
entrance animation on every rerun, and this block redraws on every send.
"""
items = _transcript_items(state)
if not items:
return
theme.section_label("Conversation so far")
theme.evidence_list(items)
# ---------------------------------------------------------------------------
# Start over
# ---------------------------------------------------------------------------
#
# One click used to be enough to delete a whole bank of answers — up to 63 of
# them, typed by hand, with nothing that could bring them back. So the first
# click only asks, and the second one, which says what it does, is the one
# that clears.
def _restart_click(target: Any, key: str, *, use_container_width: bool = False) -> str:
"""Draw "Start over" as a two-step confirmation.
Returns the click this run produced — ``"ask"``, ``"confirm"``,
``"cancel"`` or ``""``. ``target`` is whatever the buttons should be drawn
into (``st`` or a column). Nothing is decided here: the caller acts once
its card is finished, the way every other click on this screen is handled.
"""
if not st.session_state.get(_RESTART_CONFIRM_KEY):
if target.button(
"Start over",
key=key,
use_container_width=use_container_width,
help="Clear every answer and begin again. You'll be asked to confirm.",
):
return "ask"
return ""
clicked = ""
if target.button(
"Yes, delete my answers",
key=f"{key}_confirm",
use_container_width=use_container_width,
help="This cannot be undone.",
):
clicked = "confirm"
if target.button(
"Keep my answers",
key=f"{key}_cancel",
use_container_width=use_container_width,
help="Leave the interview exactly as it is.",
):
clicked = clicked or "cancel"
return clicked
def _restart_warning(state: Any) -> None:
"""What the confirm click is about to cost, in answers."""
answered = _answer_count(state)
built = st.session_state.get(_SOURCE_KEY) == "interview"
if answered:
st.warning(
f"Start over deletes {answered} answer"
+ ("" if answered == 1 else "s")
+ (" and the profile built from them," if built else "")
+ " and begins again from the first question. This can't be undone."
)
else:
st.warning(
"Start over begins again from the first question. This can't be "
"undone."
)
def _chosen_mode(settings: Any, ai_on: bool) -> tuple[str, str]:
"""The AI mode to start in, plus what to say when it can't be the one asked for.
This is where the live probe happens: the user has chosen to start, so a
network call is something they asked for rather than the price of opening
the screen. :func:`_ai_ready` caches the verdict, so the reruns that carry
the conversation never repeat it.
"""
if not ai_on:
return "off", ""
# A failed check is asked again when the user starts or resumes: that is
# their decision, and a dropped connection an hour ago should not keep
# the AI off for the rest of the session.
if _checked() and not _cached_ai_ready():
st.session_state.pop(_API_CHECK_KEY, None)
if _ai_ready(settings):
return "on", str(st.session_state.get(_ROUTE_NOTE_KEY, "") or "")
return "off", _ai_off_reason()
def _drop_interview_profile() -> None:
"""Take a deleted interview's answers out of the profile the app is using.
A folder read before a gap-filling interview is rebuilt on its own;
otherwise the profile is cleared, which locks Job Screenshot again until
a new one is built.
"""
ss = st.session_state
if ss.get(_SOURCE_KEY) != "interview":
return
base = [c for c in (ss.get("dossier_chunks") or []) if not _is_interview_chunk(c)]
if base:
try:
proofs, profile, meta = build_evidence_index(base)
except Exception: # noqa: BLE001 - fall through to a clean slate
proofs, profile, meta = [], None, {}
ss["dossier_chunks"] = base
ss["evidence_index"] = proofs or None
ss["canonical_profile"] = profile if proofs else None
ss["evidence_index_meta"] = meta if proofs else None
ss["dossier_read"] = bool(proofs)
ss[_SOURCE_KEY] = "dossier"
else:
for key in ("dossier_chunks", "evidence_index", "canonical_profile", "evidence_index_meta"):
ss[key] = None
ss["dossier_read"] = False
ss[_SOURCE_KEY] = None
ss[_STRENGTH_KEY] = None
ss["extracted_job_fields"] = None
clear_opportunity_state(ss)
def _apply_restart(click: str, *, settings: Any, ai_on: bool) -> None:
"""Act on a "Start over" click, then rerun."""
if click == "ask":
st.session_state[_RESTART_CONFIRM_KEY] = True
elif click == "cancel":
st.session_state[_RESTART_CONFIRM_KEY] = False
elif click == "confirm":
_drop_interview_profile()
mode, ack = _chosen_mode(settings, ai_on)
_start(mode, ack=ack) # clears the confirm step itself
else:
return
st.rerun()
# ---------------------------------------------------------------------------
# Cards
# ---------------------------------------------------------------------------
def _ai_choice(state: Any, *, ai_ready: bool) -> bool:
"""The AI toggle's value: what the user last chose, not a fresh default.
Read from a plain session key — and from the interview state behind it —
because the toggle's own key is collected while the interview runs. An
explicit "off" stays off across that gap, and an explicit "on" stays on.
"""
if not ai_ready:
return False
stored = st.session_state.get(_AI_CHOICE_KEY)
if isinstance(stored, bool):
return stored
if state is not None:
return str(getattr(state, "ai_mode", "") or "").strip().lower() == "on"
return True
def _card_intro(settings: Any) -> None:
"""What the interview is, and the one button that starts it."""
state = st.session_state.get(_STATE_KEY)
total = len(_questions())
# Nobody has probed the provider yet unless an earlier start did: opening
# this card is not a decision, so it must not spend a call finding out.
# The AI is offered whenever there is a key to try it with — the probe
# runs on the way in to the first question, and a failed one is asked
# again on every Start or Resume (see :func:`_chosen_mode`), so a
# provider that was down earlier in the session does not lock the toggle
# off for good.
checked = _checked()
# With no key at all the answer is already known without a probe.
has_key = _has_interview_key(settings)
ai_ready = has_key
with st.container(border=True, key="ups_panel_interview_1"):
theme.section_label("Answer questions instead of uploading files")
st.markdown(
"The app asks about your work — what you do, who you do it for, "
"and the results you have delivered. Your answers become the same "
"profile evidence a folder of files would produce, and every "
"proposal is grounded in them."
)
st.markdown(
"**Your answers are saved exactly as you type them** — word for "
"word, nothing rewritten. The AI reads them to check an answer "
"fits the question, to ask a follow-up when something is "
"missing, and to keep the conversation moving."
)
if total:
st.caption(
f"{total} questions, grouped into short sections. You can skip "
"any question and finish whenever you like."
)
# Set through session state, before the widget exists this run, only
# when a key was just accepted; otherwise seeded by ``value=`` when
# the widget is new. Never both, which Streamlit warns about.
toggle_seed: dict = {}
if st.session_state.pop(_AI_FORCE_ON_KEY, False) and ai_ready:
st.session_state[_AI_TOGGLE_KEY] = True
elif _AI_TOGGLE_KEY not in st.session_state:
toggle_seed["value"] = _ai_choice(state, ai_ready=ai_ready)
ai_on = st.toggle(
"Let the AI guide the interview",
key=_AI_TOGGLE_KEY,
disabled=not ai_ready,
**toggle_seed,
help=(
"The AI checks each answer fits the question, asks the odd "
"follow-up, and phrases the next question. Turn it off to be "
"asked the questions as written. Your answers are stored as "
"you typed them either way."
),
)
# Remembered outside the widget, because the widget's key does not
# survive the reruns that draw the question card instead of this one.
# A disabled toggle reads False whatever the user meant, so having no
# key to try is not allowed to overwrite their choice.
if ai_ready:
st.session_state[_AI_CHOICE_KEY] = bool(ai_on)
if checked and not _cached_ai_ready():
st.caption(_ai_off_reason())
if has_key:
st.caption(
"It is checked again each time you start or resume with "
"the AI on."
)
elif not has_key:
st.caption(
"No API key is available for the interview, so the AI is off. "
"Add a free Groq key below to turn it on — or start without "
"it and add one any time during the interview. "
+ _ANSWERS_VERBATIM
)
else:
name, is_setup = _route_name(settings)
if name:
st.caption(
f"The AI runs on your {name} API from Setup."
if is_setup
else f"The AI runs on your {name} key, so the interview "
"doesn't use your main API."
)
if not checked:
st.caption(
"It is checked once, when you start. If it can't be "
"reached the interview carries on without it."
)
if st.session_state.get(_ROUTE_NOTE_KEY):
st.caption(_plain(st.session_state[_ROUTE_NOTE_KEY]))
_key_form(settings, state, where="intro")
col_start, col_back = st.columns([2, 1])
start_clicked = col_start.button(
"Resume the interview" if state is not None else "Start the interview",
type="primary",
key="interview_start_btn",
use_container_width=True,
)
back_clicked = col_back.button(
"Back to Dossier",
key="back_to_dossier_from_interview",
use_container_width=True,
)
restart_click = ""
if state is not None:
if st.session_state.get(_RESTART_CONFIRM_KEY):
_restart_warning(state)
restart_click = _restart_click(st, "interview_restart_btn")
if start_clicked:
mode, ack = _chosen_mode(settings, bool(ai_on))
if state is None:
_start(mode, ack=ack)
else:
state.ai_mode = mode
state.route_pref = str(st.session_state.get(_ROUTE_PREF_KEY, "") or "")
if mode == "on":
state.ai_off_reason = ""
state.ai_off_status = ""
state.soft_failures = 0
st.session_state[_STARTED_KEY] = True
st.session_state[_RESTART_CONFIRM_KEY] = False
_resume_turn_display(state, ack)
st.rerun()
if restart_click:
_apply_restart(restart_click, settings=settings, ai_on=bool(ai_on))
if back_clicked:
st.session_state.current_step = "dossier"
st.rerun()
def _card_question(state: Any, question: Any, index: int, *, debug: bool) -> None:
"""The live question, its options, the answer box and the turn controls."""
questions = _questions()
total = len(questions)
answered = _answer_count(state)
last = st.session_state.get(_LAST_TURN_KEY) or {}
kind = str(getattr(question, "kind", "text") or "text")
options = [str(o) for o in (getattr(question, "options", None) or [])]
awaiting_reply = str(last.get("kind", "")) in ("retry", "follow_up")
picked: list[str] = []
with st.container(border=True, key="ups_panel_interview_2"):
theme.section_label(str(getattr(question, "section", "") or "Interview"))
theme.html(theme.meter(100.0 * index / total if total else 0.0))
st.caption(
f"Question {min(index + 1, total)} of {total} · {answered} answered"
)
# Always on screen, so whether the AI is judging the answers — and
# which one — is never a guess: the faint note that used to say so
# was easy to miss.
if str(getattr(state, "ai_mode", "") or "") == "on":
name, _ = _route_name(get_settings())
theme.html(
theme.tag(f"AI check on · {name or 'AI'} reviews every answer", "ok")
)
else:
theme.html(
theme.tag("AI check off · basic checks only", "warn")
)
if st.session_state.get(_ROUTE_PREF_KEY) == "main" and st.session_state.get(_ROUTE_NOTE_KEY):
st.caption(_plain(st.session_state[_ROUTE_NOTE_KEY]))
if last.get("ack"):
st.caption(_plain(last["ack"]))
# The service's reply is model-written, so it goes through a plain
# Streamlit element — never through a theme helper that takes markup —
# and is escaped, so "$10 or $20" reads as money, not maths.
if awaiting_reply and last.get("message"):
st.info(_plain(last["message"]))
st.caption(f"Original question: {_plain(getattr(question, 'text', ''))}")
else:
st.markdown(f"**{_plain(getattr(question, 'text', ''))}**")
if last.get("note"):
st.caption(_plain(last["note"]))
# After Previous, the earlier answer comes back to be edited.
editing = getattr(state, "editing", None) == index
earlier = str((getattr(state, "answers", {}) or {}).get(index, "") or "") if editing else ""
if editing and earlier and kind in ("chips", "multi"):
st.caption(f"Your earlier answer: {_plain(earlier)}")
if bool(getattr(question, "self_proof", False)):
st.caption(
"Be specific here — this answer becomes proof the proposals quote."
)
# The sequence counter is part of every answer widget's key, so the
# box and the picked options re-mount empty once a turn has consumed
# them — and keep what is in them through a re-ask, the one turn that
# asks the user to improve what they just sent.
seq = int(st.session_state.get(_SEQ_KEY, 0))
# Option buttons send straight away: one click, one turn. They stay on
# screen through a re-ask, because a re-ask on an options question is
# the service asking the user to pick one of them.
if options and kind in ("chips", "multi"):
if kind == "chips":
cols = st.columns(min(3, len(options)))
for position, option in enumerate(options):
if cols[position % len(cols)].button(
option,
key=f"iv_chip_{index}_{position}",
use_container_width=True,
):
_submit(state, option, from_chip=True)
else:
# Typing a name that is not in the list adds it as a new
# option, so a sector the list misses can be picked here too.
picked = st.multiselect(
"Pick everything that applies",
options,
key=f"iv_multi_{index}_{seq}",
accept_new_options=True,
placeholder="Pick from the list, or type your own and press Enter",
)
# On a pick-several question the box is the "not in the list" half of
# the answer, so it carries a visible label saying so; elsewhere it is
# the answer itself and the question above is its label.
multi_box = bool(options) and kind == "multi"
# An answer and its follow-up replies are stored joined, so the one
# coming back to be edited can run past the limit for one answer.
# The box holds all of it: Streamlit cuts a box's value to
# ``max_chars`` out of sight, and Send would then store what was left.
box_limit = _MAX_ANSWER_CHARS
if editing and kind == "text":
box_limit = max(box_limit, len(earlier))
text = st.text_area(
"Not in the list? Type it here" if multi_box else "Your answer",
key=f"iv_answer_{index}_{seq}",
value=earlier if (editing and kind == "text") else "",
placeholder=str(
getattr(question, "placeholder", "") or "Type your answer…"
),
height=110,
max_chars=box_limit,
label_visibility="visible" if multi_box else "collapsed",
)
if box_limit > _MAX_ANSWER_CHARS:
st.caption(
f"Your earlier answer, with what you added after a follow-up, is "
f"longer than the {_MAX_ANSWER_CHARS:,} characters one answer "
f"can hold. Skip "
f"keeps it as it is; to change it, shorten it to "
f"{_MAX_ANSWER_CHARS:,} characters first."
)
col_send, col_prev, col_skip, col_finish = st.columns([2, 1, 1, 1])
send_clicked = col_send.button(
"Send",
type="primary",
key="iv_send_btn",
use_container_width=True,
)
can_go_back = index > 0
prev_clicked = col_prev.button(
"Previous",
key="iv_prev_btn",
use_container_width=True,
disabled=not can_go_back,
help=(
"Go back to the question before this one to change your "
"answer. Sending replaces it; Skip keeps it."
if can_go_back
else "This is the first question."
),
)
skip_clicked = col_skip.button(
"Skip",
key="iv_skip_btn",
use_container_width=True,
help="Move on without answering this one.",
)
can_finish = answered >= _MIN_ANSWERS_TO_FINISH
finish_clicked = col_finish.button(
"Finish",
key="iv_finish_early_btn",
use_container_width=True,
disabled=not can_finish,
help=(
"Build my profile from what I have answered so far."
if can_finish
else f"Answer at least {_MIN_ANSWERS_TO_FINISH} questions first."
),
)
if debug:
st.caption(
f"use: {getattr(question, 'use', '—')} · kind: {kind} · "
f"AI turns used: {getattr(state, 'calls_used', 0)} of "
f"{getattr(interview, 'MAX_CALLS_PER_INTERVIEW', '—')} · "
f"mode: {getattr(state, 'ai_mode', '—')}"
)
# Below the controls, so the answer box stays where the eye expects
# it; shown only while the AI is off for want of a working key.
if str(getattr(state, "ai_mode", "") or "") != "on":
st.divider()
_key_form(get_settings(), state, where="question")
else:
_show_key_flash()
# Clicks are read first and acted on afterwards, so every widget in
# this card has already handed over its value.
if send_clicked:
typed = (text or "").strip()
# A "multi" question asks the user to pick all that apply *or*
# add their own, so the ticks and the typed words are two halves
# of one answer. Sending either half on its own would silently
# throw the other away.
answer = ", ".join([*picked, typed]) if typed else ", ".join(picked)
# An answer made only of option picks is the same thing a chip
# button sends: a choice from a fixed list the model never has
# to read. Sent as a chip it costs no call against the free tier.
# Typed words are not on that list, so as soon as there are any
# the whole answer goes through validation.
# A name typed into the dropdown is a new option, not one of the
# list's, so it goes through validation like typed words do.
new_picks = [p for p in picked if p not in options]
from_picks = bool(picked) and not typed and not new_picks
if not answer:
st.warning("Type an answer first, or use Skip.")
elif len(typed) > _MAX_ANSWER_CHARS:
# Only the long earlier answer above gets this far, and the
# service keeps the first ``_MAX_ANSWER_CHARS`` of an answer:
# sent, it would lose its end without a word.
st.warning(
f"That answer is longer than the {_MAX_ANSWER_CHARS:,} "
"characters one answer can hold, so sending it would cut "
"off the end. Shorten it first, or use Skip to keep your "
"earlier answer as it is."
)
else:
_submit(state, answer, from_chip=from_picks)
if prev_clicked:
_previous(state)
if skip_clicked:
_skip(state)
if finish_clicked:
# An answer typed or ticked but not sent would be lost without a word.
# After Previous the box holds the stored answer; that one is saved.
unsent = (text or "").strip()
if editing and unsent == earlier.strip():
unsent = ""
if unsent or picked:
st.warning(
"You have an answer here that isn't sent yet. Send it (or clear "
"the box) first, then Finish."
)
else:
_finish(state)
def _card_profile(state: Any, *, debug: bool) -> None:
"""The profile as it stands, so the user sees the interview paying off."""
with st.container(border=True, key="ups_panel_interview_3"):
theme.section_label("Your profile so far")
strength = _strength(state)
score = int(strength.get("score", 0) or 0)
label = str(strength.get("label", "") or "")
theme.kpis(
[
{
"lab": "Profile strength",
"val": f"{score}/100",
"icon": "spark",
"delta": label,
"delta_kind": _strength_tone(score),
}
],
columns=1,
)
theme.html(theme.meter(score, _strength_band(score)))
pairs = _profile_pairs(state)
if pairs:
theme.kv_grid(pairs, columns=2)
else:
st.caption("Answer a few questions and your profile fills in here.")
if debug:
breakdown = strength.get("breakdown") or {}
if isinstance(breakdown, dict) and breakdown:
for key, value in breakdown.items():
st.write(f"- {key}: **{value}**")
def _card_finish(state: Any, settings: Any) -> None:
"""The end of the bank: build the profile and move on.
Build is gated by the same minimum the in-question Finish button uses.
Reaching the end of the bank is not the same as having answered it: a
user who skipped every question arrives here with nothing to build
from, and an ungated button would only hand them an error. So the
button waits, and the card says what to do instead — including the way
back to the folder route, which is the only other way out of here.
"""
answered = _answer_count(state)
can_build = answered >= _MIN_ANSWERS_TO_FINISH
with st.container(border=True, key="ups_panel_interview_4"):
theme.section_label("Interview complete")
if can_build:
st.success(f"That's everything — {answered} answers recorded.")
st.write(
"Next the app turns your answers into proof points, the same "
"way it does with a folder of files."
)
else:
st.warning(
(
"That's the end of the questions, but none of them were "
"answered"
if answered == 0
else f"That's the end of the questions, but only "
f"{answered} were answered"
)
+ f" — fewer than the {_MIN_ANSWERS_TO_FINISH} a profile "
"needs. Go back to the questions and answer a few, or build "
"your profile from a folder of files instead."
)
if st.session_state.get(_RESTART_CONFIRM_KEY):
_restart_warning(state)
col_build, col_questions, col_restart, col_back = st.columns([2, 1, 1, 1])
questions_clicked = col_questions.button(
"Back to the questions",
key="iv_back_to_questions_btn",
use_container_width=True,
help="Return to the last question. Use Previous to reach any you skipped.",
)
build_clicked = col_build.button(
"Build my profile and continue",
type="primary",
key="iv_finish_btn",
use_container_width=True,
disabled=not can_build,
help=(
"Build my profile from what I have answered so far."
if can_build
else f"Answer at least {_MIN_ANSWERS_TO_FINISH} questions first."
),
)
restart_click = _restart_click(
col_restart, "iv_restart_from_done_btn", use_container_width=True
)
back_clicked = col_back.button(
"Back to Dossier",
key="back_to_dossier_from_done_interview",
use_container_width=True,
help="Build your profile from a folder of files instead.",
)
if build_clicked:
_finish(state)
if questions_clicked:
_previous(state)
if restart_click:
# The user's own choice, not the state's mode: an interview whose AI
# dropped out mid-way should get a fresh check, not stay off.
choice = st.session_state.get(_AI_CHOICE_KEY)
_apply_restart(
restart_click,
settings=settings,
ai_on=choice if isinstance(choice, bool) else True,
)
if back_clicked:
st.session_state.current_step = "dossier"
st.rerun()
# ---------------------------------------------------------------------------
# Public render
# ---------------------------------------------------------------------------
def render() -> None:
if not st.session_state.get("api_ok"):
st.error("This step is locked. Run the API check on the Setup screen first.")
if st.button("Back to Setup", key="back_to_setup_from_interview"):
st.session_state.current_step = "setup"
st.rerun()
return
settings = get_settings()
debug = bool(getattr(settings, "show_debug_panel", False))
theme.screen_head(
"dossier",
"Let's build your profile",
"No folder of files? Answer questions instead — the app builds the "
"same evidence from <b>your own words</b> and grounds every proposal "
"in it.",
)
if not is_available() or not _questions():
st.error(
"The interview isn't available right now. You can still build "
"your profile from a folder of files."
)
if st.button("Back to Dossier", key="back_to_dossier_unavailable"):
st.session_state.current_step = "dossier"
st.rerun()
return
# No provider probe here. Arriving on this screen is not a decision, and
# a live call fired before the user has chosen anything spends their free
# quota on a screen they may be about to leave. The one check runs when
# they press Start (see :func:`_chosen_mode`) and is cached for the rest
# of the session, so the reruns that carry the conversation never repeat
# it. Its verdict decides whether the AI turns run and what the user is
# told; it never stops the interview.
state = st.session_state.get(_STATE_KEY)
if state is None or not st.session_state.get(_STARTED_KEY):
_card_intro(settings)
return
if _is_done(state):
_card_finish(state, settings)
_card_profile(state, debug=debug)
_render_transcript(state)
return
question, index = _current_question(state)
if question is None:
# Defensive: an index past the bank means the interview is over.
state.done = True
_card_finish(state, settings)
return
col_chat, col_profile = st.columns([3, 2])
with col_chat:
_card_question(state, question, index, debug=debug)
with col_profile:
_card_profile(state, debug=debug)
with col_chat:
_render_transcript(state)
if st.button("Back to Dossier", key="back_to_dossier_from_running_interview"):
st.session_state.current_step = "dossier"
st.rerun()
"""Interview screen — build the profile by conversation instead of by files.
A user with no dossier folder (or a thin one) answers a guided set of
questions here and ends up with the same profile evidence the folder path
produces. The screen owns the conversation only: the question bank, the
answer validation, the follow-ups and the evidence build all live in
:mod:`app.services.interview`.
Answers are stored exactly as the user typed them: nothing here rewrites,
tidies or "cleans" what they wrote. The AI earns its place on the
conversation instead — checking an answer fits the question that was asked,
asking a follow-up when a needed detail is missing, phrasing the next
question, and acknowledging what was said.
On completion this writes exactly the session state a dossier read writes —
``dossier_chunks``, ``dossier_read``, ``dossier_read_error``,
``evidence_index``, ``canonical_profile``, ``evidence_index_meta`` — so every
screen downstream of Dossier works unchanged, plus ``profile_source`` so the
rest of the app can tell the two paths apart.
The interview is a second way through the Dossier step rather than a step of
its own: it renders under the Dossier rail row and the Dossier step number
(see ``STEP_PARENT`` in ``app/main.py``).
Nothing here is imported at provider level: the screen renders a clean
message and stays out of the way if the interview service is unavailable.
"""
from __future__ import annotations
import traceback
import copy
import re
from types import SimpleNamespace
from typing import Any
import streamlit as st
from app.config import clear_opportunity_state, get_settings
from app.services.evidence_index import build_evidence_index
from app.ui import theme
from app.utils.logging_utils import get_logger
_logger = get_logger("upwork_strategist.interview_screen")
try: # The interview service is a separate module; a missing or broken
# import must never take the rest of the app down with it.
from app.services import interview
except Exception: # noqa: BLE001 - the screen degrades to a clean message
interview = None # type: ignore[assignment]
# ---------------------------------------------------------------------------
# Session keys
# ---------------------------------------------------------------------------
#
# The first three are the contract keys other screens read. The last two are
# display-only state owned by this screen: the latest turn's reply, and a
# counter that forms part of the answer box's widget key so the box re-mounts
# empty after every send (the widget-reset idiom used elsewhere in this app).
_STATE_KEY = "interview_state"
_STARTED_KEY = "interview_started"
_SOURCE_KEY = "profile_source"
_LAST_TURN_KEY = "interview_last_turn"
_SEQ_KEY = "interview_turn_seq"
# The AI toggle's value, kept outside the widget. The intro card is not drawn
# while the interview runs, so Streamlit collects the toggle's own key, and a
# toggle seeded from ``value=`` alone springs back to its default the next
# time the card is drawn — silently overriding a choice the user made on
# purpose. This key is a plain value and survives that gap.
_AI_CHOICE_KEY = "interview_ai_choice"
_AI_TOGGLE_KEY = "interview_ai_toggle"
# Set when a Groq key is accepted on this screen. Streamlit keeps a widget's
# last value and ignores a new ``value=``, so without this the toggle would
# stay off — drawn disabled while there was no key — and the user who was
# just told "AI help is on" would start an interview without it.
_AI_FORCE_ON_KEY = "interview_ai_force_on"
# "Start over" throws away every answer given so far, so it asks first: this
# holds the "asked, not yet confirmed" step between the two clicks.
_RESTART_CONFIRM_KEY = "interview_restart_confirm"
# Below this many answered questions there is not enough to build a profile
# from, so "Finish early" stays disabled.
_MIN_ANSWERS_TO_FINISH = 3
# What a skip and a step backwards say. The skip line only tells the truth
# because "Previous" exists: without a way back there would be no "later".
_SKIP_ACK = "Skipped — you can come back to it with Previous."
_BACK_ACK = (
"Back to the previous question. Your earlier answer is in the box: edit it "
"and Send to replace it, or Skip to keep it."
)
# Which API the interview runs on this session: "" tries a Groq key first,
# "main" is set when the Groq check failed and the Setup API took over, and
# the note says so in words.