"""Core fidelity checks for Discovery questions and invented constraints. These rules do not admit capabilities, change Planner matching, or alter Phase 12 contracts. They only decide whether a first-turn question or high-impact rule is already answered by the user or is an invented consequential constraint. """ from __future__ import annotations import re from src.domain.projects.discovery_schemas import ( Actor, AuthenticationSpec, DiscoveryQuestion, DiscoverySpec, DiscoveryTurnResult, IntegrationSpec, RequestedCapability, Workflow, WorkflowStep, ) # These are customer-language signals, not platform/API vocabulary. Discovery must translate # them into the internal capability contract before Codegen sees the plan. Keep the vocabulary # deliberately conservative: staff-only access must not silently become customer auth. CUSTOMER_AUTH_MARKERS = ( "sign up", "signup", "register", "create an account", "create account", "customer login", "customer sign in", "customer signin", "member login", "member sign in", "user account", "user accounts", "customer account", "customer accounts", "member account", "member accounts", "user profile", "customer profile", ) GENERIC_ACCOUNT_AUTH_MARKERS = ( "users sign in", "users signin", "users log in", "users login", "members sign in", "members log in", "members login", "people sign in", "people log in", "customer portal", "member portal", "user portal", ) PASSWORD_AUTH_MARKERS = ( "password", "password-based", "email and password", "email/password", ) MAGIC_LINK_AUTH_MARKERS = ( "magic link", "passwordless", "one-time link", "one time link", ) SIGNUP_MARKERS = ( "sign up", "signup", "register", "create an account", "create account", ) GENERIC_LOGIN_RE = re.compile(r"\b(sign\s*in|signin|log\s*in|login)\b", re.I) def normalize_capability_intent(spec: DiscoverySpec, payload: dict) -> DiscoverySpec: """Translate explicit customer language into Core's capability contract. The Discovery model is intentionally allowed to omit derived enum fields. Codegen, however, must receive a deterministic contract. This function fills only the auth facts supported by the user's words; it never derives auth from generated files or relaxes conformance checks. """ text = user_message_text(payload) lowered = text.lower() customer_auth = any(marker in lowered for marker in CUSTOMER_AUTH_MARKERS) customer_auth = customer_auth or any(marker in lowered for marker in GENERIC_ACCOUNT_AUTH_MARKERS) customer_auth = customer_auth or ( bool(re.search(r"\b(customer|customers|member|members|user|users)\b", lowered)) and any(marker in lowered for marker in (*MAGIC_LINK_AUTH_MARKERS, *PASSWORD_AUTH_MARKERS)) ) # A bare "login" is ambiguous (it can mean staff access). Treat it as customer auth only # when the surrounding request names end users/customers/members, or asks for signup. if not customer_auth and GENERIC_LOGIN_RE.search(lowered): customer_auth = bool( re.search(r"\b(customer|customers|user|users|member|members|visitor|visitors)\b", lowered) ) and not any(marker in lowered for marker in STAFF_ACCESS_MARKERS) if not customer_auth: return spec existing = spec.authentication methods = list(existing.methods if existing else []) audience_methods = { key: list(value) for key, value in (existing.audience_methods.items() if existing else []) } explicit_password = any(marker in lowered for marker in PASSWORD_AUTH_MARKERS) explicit_magic_link = any(marker in lowered for marker in MAGIC_LINK_AUTH_MARKERS) if explicit_password: methods.append("password") if explicit_magic_link: methods.append("magic_link") # If the customer says users can log in but does not choose a method, admit the two # installed conventional methods. Codegen may choose the UX; it may not invent an API. if not explicit_password and not explicit_magic_link: methods.extend(("magic_link", "password")) methods = list(dict.fromkeys(methods)) audiences = list(existing.audiences if existing else []) if "customer" not in audiences: audiences.append("customer") customer_methods = list(dict.fromkeys([*audience_methods.get("customer", []), *methods])) audience_methods["customer"] = customer_methods signup_audiences = list(existing.signup_audiences if existing else []) if any(marker in lowered for marker in SIGNUP_MARKERS): if "customer" not in signup_audiences: signup_audiences.append("customer") normalized = AuthenticationSpec( required=True, methods=methods, audiences=audiences, audience_methods=audience_methods, signup_audiences=signup_audiences, notes=existing.notes if existing else "Customer authentication inferred from the user's request.", ) return spec.model_copy(update={"authentication": normalized}) STAFF_ACCESS_MARKERS = ( "staff-only", "staff only", "team-only", "team only", "internal-only", "internal only", "staff must log in", "staff must", "staff need", "staff can", "logged-in staff", "logged in staff", "employees sign in", "employees can", "staff review", ) PUBLIC_SUBMIT_MARKERS = ( "public", "visitors", "contact page", "request-a-quote", "request a quote", "quote form", ) EXTERNAL_FORM_MARKERS = ( "jotform", "typeform", "google forms", "microsoft forms", "hubspot form", "external form", "third-party form", "third party form", "embedded form", "embed form", "form embed", "iframe form", ) PAYMENTS_EXCLUDED_MARKERS = ( "no online payment", "no online payments", "no payment", "no payments", "no stripe", "no billing", "billing later", ) ACCESS_QUESTION_MARKERS = ( "access role", "access roles", "who will have access", "who can access", "different roles", "permissions for staff", "staff members in the app", "admin", "editor", "viewer", "user roles", "roles or permissions", "access levels for staff", "how will staff access", "how should staff access", "how will employee access", ) CODEGEN_DETAIL_MARKERS = ( "specific field", "which fields", "form fields", "short form", "thank-you", "thank you message", "authentication method", "auth method", "login method", "which pages", "page list", "site map", "homepage structure", "section order", "component", "layout", "color", "font", "visual style", "brand identity", "data fields", "what specific details about each customer", "what specific details about the customers", "what specific customer details", "what specific information should be collected", "what specific information should the public form collect", "what specific information should be included in the request", "what specific details should employees", # Conventional form-field questions Codegen owns (user already named the forms). "what information should be collected", "what fields should be included", "information should be collected in the", "fields should be included in the contact", # Marketing/portfolio UX — navigation and content engagement are Codegen defaults. "main user interactions", "user interactions or flows", "navigation through sections", "engagement with content", "how visitors navigate", "how should visitors browse", "browse the portfolio", "navigate the portfolio", "interact with the portfolio", "flows you envision for the portfolio", "sections of the site", "mobile responsive", "seo ", "analytics", ) PAYMENT_QUESTION_MARKERS = ("payment", "billing", "stripe", "subscription") CONVENTIONAL_CONSTRAINT_MARKERS = ( "unique identifier", "unique id", "must log in", "staff must", "timestamp", "validation", "empty state", "password reset", ) INVENTED_CONSTRAINT_MARKERS = ( "one at a time", "only one", "one inquiry", "unique product name", "unique names", "unique name", "must have unique name", "creator-only", "only the creator", "only its creator", "can only edit", "they have created", "cannot be edited", "permission-dependent", ) INVENTED_ROLE_SCOPES = ( "varies based on role", "varies by role", "depends on role", "role-based", "admin/editor/viewer", ) ROLE_LADDER_NAMES = frozenset( {"admin", "editor", "viewer", "administrator", "manager", "supervisor"} ) # Generic, non-hierarchical worker terms a role ladder can collapse back down to. Deliberately # limited to common English words for "an undifferentiated group of workers" — never a # business-specific actor name (e.g. "Purchasing Staff" is not in this set and must not collapse). BASE_WORKER_NAMES = frozenset({"staff", "employee", "employees", "user", "member"}) def user_message_text(payload: dict) -> str: parts = [] latest = payload.get("latest_input") or {} if isinstance(latest, dict) and latest.get("content"): parts.append(str(latest["content"])) for message in payload.get("messages") or payload.get("conversation") or []: if message.get("role") == "user" and message.get("content"): parts.append(str(message["content"])) return "\n".join(parts) def user_facts(text: str) -> frozenset[str]: lowered = text.lower() facts: set[str] = set() if any(marker in lowered for marker in STAFF_ACCESS_MARKERS): facts.add("staff_access") if any(marker in lowered for marker in PUBLIC_SUBMIT_MARKERS): facts.add("public_submit") if any(marker in lowered for marker in PAYMENTS_EXCLUDED_MARKERS): facts.add("payments_excluded") if "short form" in lowered or "form" in lowered: facts.add("conventional_form") if any(marker in lowered for marker in EXTERNAL_FORM_MARKERS): facts.add("external_form") return frozenset(facts) def has_explicit_access_policy(text: str) -> bool: return bool(re.search(r"\b(admin|administrator|editor|viewer|manager)\b", text, re.I)) or any( marker in text.lower() for marker in ("their own", "assigned to", "approval", "approve", "only managers") ) def question_is_unnecessary( question: DiscoveryQuestion | None, spec: DiscoverySpec, payload: dict ) -> bool: if question is None: return False text = user_message_text(payload) if not text.strip(): return False facts = user_facts(text) # Options can mention a role or layout without that being the actual question. asked = question.text.lower() if any( word in asked for word in ( "approval", "approve", "tenant", "companies", "organizations", "own records", "other users", "other employees", "payment provider", ) ): return False if has_explicit_access_policy(text) and any( word in asked for word in ("role", "permission", "access") ): return False if any(marker in asked for marker in ACCESS_QUESTION_MARKERS) and "staff_access" in facts: return True if any(marker in asked for marker in CODEGEN_DETAIL_MARKERS): return True if "external_form" in facts and any( marker in asked for marker in ( "form", "fields", "field", "actions", "workflow", "submission", "what should happen", ) ): return True if any(marker in asked for marker in PAYMENT_QUESTION_MARKERS) and "payments_excluded" in facts: return True return False def is_invented_constraint(statement: str, _impact: str, user_text: str) -> bool: lowered = statement.lower() source = user_text.lower() if any(marker in lowered for marker in CONVENTIONAL_CONSTRAINT_MARKERS): return False matched = [marker for marker in INVENTED_CONSTRAINT_MARKERS if marker in lowered] if not matched: return False if any(marker in source for marker in matched): return False return True def invented_constraints(spec: DiscoverySpec, payload: dict) -> list[str]: text = user_message_text(payload) found: list[str] = [] for rule in spec.business_rules: if is_invented_constraint(rule.rule, rule.impact, text): found.append(rule.rule) for assumption in spec.assumptions: if is_invented_constraint(assumption.statement, assumption.impact, text): found.append(assumption.statement) return found def _looks_like_invented_role_split(scope: str) -> bool: return any(marker in scope.lower() for marker in INVENTED_ROLE_SCOPES) def _ungrounded_role_ladder(actors: list[Actor], text: str) -> bool: """True when the actor set is a hierarchy-tier ladder (admin/manager/editor/...) over a generic worker base, with no textual grounding anywhere in the user's own message for that specific hierarchy. Independent of any staff-only access phrasing — a prompt that never mentions access at all still qualifies, since the invented tiers have no support either way. Never fires for a business-specific actor name (e.g. "Purchasing Staff"), only for the closed, generic vocabulary in ROLE_LADDER_NAMES/BASE_WORKER_NAMES. """ if len(actors) < 2 or has_explicit_access_policy(text): return False names = {actor.name.strip().lower() for actor in actors} return bool(names & ROLE_LADDER_NAMES) and names <= (ROLE_LADDER_NAMES | BASE_WORKER_NAMES) def apply_fidelity_hygiene(spec: DiscoverySpec, payload: dict) -> DiscoverySpec: text = user_message_text(payload) facts = user_facts(text) actors = list(spec.actors) if _ungrounded_role_ladder(actors, text): base = next( (actor for actor in actors if actor.name.strip().lower() in BASE_WORKER_NAMES), Actor(id="staff", name="Staff", description="Staff users named in the request."), ) base_label = base.name.strip().lower() actors = [ base.model_copy( update={"access_scope": f"all {base_label} records in this workspace"} ) ] elif "staff_access" in facts and not has_explicit_access_policy(text): actors = [ actor.model_copy(update={"access_scope": "all staff records in this workspace"}) if actor.name.lower() == "staff" and _looks_like_invented_role_split(actor.access_scope or "") else actor for actor in actors ] rules = [ rule for rule in spec.business_rules if not is_invented_constraint(rule.rule, rule.impact, text) ] assumptions = [ item for item in spec.assumptions if not is_invented_constraint(item.statement, item.impact, text) ] cleaned = spec.model_copy( update={"actors": actors, "business_rules": rules, "assumptions": assumptions} ) return seed_stated_form_workflows(cleaned, payload) def seed_stated_form_workflows(spec: DiscoverySpec, payload: dict) -> DiscoverySpec: """When the user already named quote/contact forms, treat them as primary workflows. Models often ask about form fields (Codegen detail) while leaving workflows empty, which Core blocks as missing_primary_workflow. Seeding from explicit user phrasing clears that false gap without inventing product scope. Marketing showcases that only say attract/showcase clients get a browse→convert workflow when no other process exists. """ text = user_message_text(payload).lower() workflows: list[Workflow] = list(spec.workflows) had_existing_workflows = bool(workflows) external_provider = next( ( provider for marker, provider in ( ("jotform", "Jotform"), ("typeform", "Typeform"), ("google forms", "Google Forms"), ("microsoft forms", "Microsoft Forms"), ) if marker in text ), None, ) external_form = bool( external_provider or any(marker in text for marker in EXTERNAL_FORM_MARKERS) ) if external_form and not any( item.id == "external_form_submission" for item in workflows ): workflows.append( Workflow( id="external_form_submission", name="External form submission", description=( "Visitors complete the named third-party form embedded or linked " "on the website." ), steps=[ WorkflowStep( name="Complete external form", description=( "Visitor uses the provider-owned form; P2HS does not recreate " "its fields or submission actions." ), ) ], outcome="The external provider receives and manages the submission.", ) ) if not had_existing_workflows and any( marker in text for marker in ( "get a quote", "quote form", "request a quote", "request quote", "quote request", ) ): workflows.append( Workflow( id="get_quote", name="Get a quote", description="Prospective clients request a quote for a service.", steps=[ WorkflowStep( name="Submit quote request", description="Visitor submits the Get a Quote form on a services page.", ) ], outcome="The team receives the quote inquiry.", ) ) if not had_existing_workflows and any( marker in text for marker in ("contact form", "contact us", "existing customer", "existing customers") ): workflows.append( Workflow( id="contact", name="Contact", description="Visitors or existing customers send a contact message.", steps=[ WorkflowStep( name="Submit contact form", description="Visitor submits the contact form.", ) ], outcome="The team receives the message.", ) ) if not workflows and _marketing_showcase_intent(text, spec): workflows.append( Workflow( id="browse_and_convert", name="Browse and inquire", description="Visitors review services or portfolio work and reach out.", steps=[ WorkflowStep( name="Browse offer", description="Visitor reviews services, work, or portfolio content.", ), WorkflowStep( name="Convert", description="Visitor contacts the business or requests a quote.", ), ], outcome="The business receives a new lead inquiry.", ) ) if not workflows: return spec forms = list(spec.forms) if external_provider and external_provider not in forms: forms.append(external_provider) if any(m in text for m in ("get a quote", "quote form", "request a quote")) and ( "Get a Quote" not in forms ): forms.append("Get a Quote") if "contact form" in text and "Contact" not in forms: forms.append("Contact") integrations = list(spec.integrations) if external_provider and not any( (item.provider or "").lower() == external_provider.lower() for item in integrations ): integrations.append( IntegrationSpec( id=external_provider.lower().replace(" ", "_"), provider=external_provider, purpose=f"Embed the customer-selected {external_provider} form on the website.", required_for_first_build=True, direction="outbound", ) ) return spec.model_copy( update={"workflows": workflows, "forms": forms[:8], "integrations": integrations[:12]} ) def _marketing_showcase_intent(text: str, spec: DiscoverySpec) -> bool: archetypes = {item.lower() for item in spec.archetypes} marketingish = not archetypes or bool( archetypes & {"marketing_site", "landing_page", "content_platform", "custom_web_app"} ) if not marketingish: return False return any( marker in text for marker in ( "showcase", "portfolio", "attract new clients", "attract clients", "web design", "web development company", "agency", "services we provide", "landing page", "marketing site", ) ) def user_stated_form_workflows(payload: dict) -> bool: text = user_message_text(payload).lower() has_quote = any( marker in text for marker in ( "get a quote", "quote form", "request a quote", "request quote", "quote request", ) ) has_contact = any( marker in text for marker in ("contact form", "contact us", "existing customer", "existing customers") ) has_external = any(marker in text for marker in EXTERNAL_FORM_MARKERS) return has_quote or has_contact or has_external def local_choose_for_me_turn(state: dict) -> DiscoveryTurnResult | None: """Resolve choose-for-me locally when only conventional form/workflow gaps remain. Avoids a second heavy provider call that often times out after several clarify turns. """ from src.domain.projects.discovery_schemas import ( DiscoveryQuestion, DiscoveryReadiness, DiscoverySpec, DiscoveryTurnResult, ) latest = state.get("latest_input") or {} if latest.get("type") != "choose_for_me": return None prev = state.get("result") or {} if not prev.get("spec") or not prev.get("question"): return None payload = { "messages": state.get("messages") or [], "latest_input": latest, "conversation": state.get("messages") or [], } question = DiscoveryQuestion.model_validate(prev["question"]) spec = DiscoverySpec.model_validate(prev["spec"]) conventional = question_is_unnecessary(question, spec, payload) or user_stated_form_workflows( payload ) if not conventional: return None return DiscoveryTurnResult( decision="clarify", message=( "I'll use a conventional setup for those form details and keep the " "quote and contact flows you already described." ), question=question, spec=spec, readiness=DiscoveryReadiness( buildable=False, blocking_decision_ids=[], reason="Resolving conventional defaults from choose-for-me.", ), requested_capabilities=list(prev.get("requested_capabilities") or []), project_type=prev.get("project_type") or "nuxt_app", template_ref=prev.get("template_ref"), template_fit=prev.get("template_fit") or "none", template_reason=prev.get("template_reason") or "", ) def should_promote_ready(turn: DiscoveryTurnResult, payload: dict, blockers: list[str]) -> bool: if turn.decision != "clarify" or blockers: return False return question_is_unnecessary(turn.question, turn.spec, payload) def reconcile_requested_capabilities(spec: DiscoverySpec, requested: list[str]) -> list[str]: """Restore capabilities the model's own already-extracted spec facts imply but its own free capability choice omitted — using the exact rule DISCOVERY_PROMPT already states ("mark auth when users sign in, records when business entities are stored"). This never adds a capability the spec doesn't already support and never removes one the model chose; it only fills a gap against facts the model itself already committed to. Reconciling against an already-stated rule is not the same as inferring product semantics: it does not decide whether entities exist or authentication is required, only keeps a derived enum field consistent with facts already on the spec. """ capabilities = set(requested) if spec.entities: capabilities.add("records") if spec.authentication and spec.authentication.required: capabilities.add("auth") return [item for item in RequestedCapability.__args__ if item in capabilities]