From 9ef6f4bd2b28f6f5e05cbb0fd91d52cc84bacaba Mon Sep 17 00:00:00 2001 From: BIN Zhang Date: Mon, 16 Mar 2026 00:22:38 +0800 Subject: [PATCH] feat: extract task dispatch registry --- demos/persona_workflow_demo.py | 440 +++++++-------------------------- demos/task_registry.py | 342 +++++++++++++++++++++++++ 2 files changed, 431 insertions(+), 351 deletions(-) create mode 100644 demos/task_registry.py diff --git a/demos/persona_workflow_demo.py b/demos/persona_workflow_demo.py index 51552bd..030ae17 100644 --- a/demos/persona_workflow_demo.py +++ b/demos/persona_workflow_demo.py @@ -1,6 +1,7 @@ from __future__ import annotations import argparse +import importlib.util import json import re import sys @@ -12,82 +13,37 @@ sys.path.insert(0, str(PROJECT_ROOT)) from demos.task_context import TaskContext -from pop.persona_loader import Persona, load_persona +from demos.task_registry import ( + build_stage_output, + deliverable_type_for, + supported_task_types, +) DEFAULT_TASK_INPUT_PATH = PROJECT_ROOT / "demos" / "market_research_task.json" DEFAULT_OUTPUT_DIR = PROJECT_ROOT / "demos" / "generated" -SUPPORTED_TASK_TYPES = { - "market_research", - "product_design", - "ux_review", -} -TASK_TYPE_PROFILES: dict[str, dict[str, Any]] = { - "market_research": { - "design_default_objective": "Design a product concept", - "research_default_objective": "Research market trends", - "marketing_default_objective": "Create marketing strategy", - "design_focus": [ - "clarify the value proposition", - "connect concept choices to buyer expectations", - "prepare a market-aware design brief", - ], - "research_focus": [ - "collect supporting market signals", - "distill concise findings", - "translate evidence into technical context", - ], - "marketing_focus": [ - "turn research into positioning", - "shape a campaign-ready narrative", - "package the final deliverable", - ], - "deliverable_type": "market_strategy_report", - }, - "product_design": { - "design_default_objective": "Shape the product concept", - "research_default_objective": "Review design constraints and competitive cues", - "marketing_default_objective": "Prepare launch messaging for the concept", - "design_focus": [ - "define the concept direction", - "prioritize user-facing differentiators", - "keep the concept buildable under constraints", - ], - "research_focus": [ - "compare adjacent product patterns", - "identify feasibility and usability constraints", - "surface practical tradeoffs for the concept", - ], - "marketing_focus": [ - "translate the concept into launch language", - "align story with buyer expectations", - "package the concept as a reviewable brief", - ], - "deliverable_type": "product_concept_brief", - }, - "ux_review": { - "design_default_objective": "Define a better checkout experience", - "research_default_objective": "Identify friction and usage patterns", - "marketing_default_objective": "Package the UX improvement plan for rollout", - "design_focus": [ - "map the intended user journey", - "reduce points of hesitation", - "prepare a concise improvement concept", - ], - "research_focus": [ - "identify friction signals", - "connect issues to user behavior patterns", - "prioritize what should change first", - ], - "marketing_focus": [ - "frame the UX work for stakeholders", - "turn findings into an adoption story", - "package a rollout-ready improvement plan", - ], - "deliverable_type": "ux_improvement_plan", - }, +STAGE_PROGRESS = { + "design": "Designing", + "research": "Researching", + "marketing": "Marketing", } +def load_persona_loader_module() -> Any: + # Load the lightweight demo loader directly to avoid package-level side effects. + module_path = PROJECT_ROOT / "pop" / "persona_loader.py" + spec = importlib.util.spec_from_file_location("persona_loader_demo", module_path) + if spec is None or spec.loader is None: + raise ImportError(f"Unable to load persona loader from {module_path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +PERSONA_LOADER_MODULE = load_persona_loader_module() +Persona = PERSONA_LOADER_MODULE.Persona +load_persona = PERSONA_LOADER_MODULE.load_persona + + def snapshot_result(result: dict[str, object]) -> dict[str, object]: return { "persona_name": result.get("persona_name"), @@ -112,10 +68,11 @@ def validate_task_input(task_input: dict[str, Any]) -> None: if not str(task_input.get("task_name", "")).strip(): raise ValueError("Task input requires a non-empty `task_name`.") task_type = str(task_input.get("task_type", "")).strip() - if task_type not in SUPPORTED_TASK_TYPES: - supported = ", ".join(sorted(SUPPORTED_TASK_TYPES)) + supported = supported_task_types() + if task_type not in supported: + supported_names = ", ".join(sorted(supported)) raise ValueError( - f"Unsupported `task_type`: {task_type or '(missing)'}. Expected one of: {supported}." + f"Unsupported `task_type`: {task_type or '(missing)'}. Expected one of: {supported_names}." ) inputs = task_input.get("inputs") if not isinstance(inputs, dict): @@ -217,217 +174,6 @@ def resolve_project_path(path: Path) -> Path: return path if path.is_absolute() else PROJECT_ROOT / path -def subject_inputs(task_input: dict[str, Any]) -> dict[str, Any]: - return task_input.get("inputs", {}) - - -def summarize_subject(task_input: dict[str, Any]) -> str: - inputs = subject_inputs(task_input) - feature_list = ", ".join(inputs.get("subject_attributes", [])) or "core attributes" - return ( - f"{inputs.get('subject_name', 'Unknown subject')} in " - f"{inputs.get('subject_category', 'Unknown')}" - f" with {feature_list}" - ) - - -def with_indefinite_article(text: str) -> str: - article = "an" if text[:1].lower() in {"a", "e", "i", "o", "u"} else "a" - return f"{article} {text}" - - -def infer_market_trends(task_input: dict[str, Any]) -> list[str]: - inputs = subject_inputs(task_input) - features = {str(feature).lower() for feature in inputs.get("subject_attributes", [])} - trends: list[str] = [] - if "5g" in features: - trends.append("5G adoption remains a purchase driver in premium devices") - if "fast charging" in features: - trends.append("Battery convenience influences upgrade decisions") - if "touchscreen" in features: - trends.append("Large responsive displays remain central to daily usage") - if str(inputs.get("subject_category", "")).lower() == "electronics": - trends.append("Consumers compare feature density with price sensitivity") - return trends or ["General market demand should be validated with recent signals"] - - -def infer_design_constraints(task_input: dict[str, Any]) -> list[str]: - inputs = subject_inputs(task_input) - constraints = inputs.get("constraints", []) - if constraints: - return constraints - return [ - "Balance concept ambition with implementation feasibility", - "Keep differentiation obvious to the target audience", - ] - - -def infer_ux_findings(task_input: dict[str, Any]) -> list[str]: - inputs = subject_inputs(task_input) - attributes = {str(item).lower() for item in inputs.get("subject_attributes", [])} - findings: list[str] = [] - if "form abandonment" in attributes: - findings.append("Long form steps likely interrupt momentum before payment.") - if "payment trust" in attributes: - findings.append("Trust cues near payment decisions need to be more visible.") - if "mobile checkout" in attributes: - findings.append("Mobile users need fewer taps and clearer progress markers.") - return findings or [ - "Review friction around the primary user path and prioritize the highest drop-off points." - ] - - -def objective_for( - task_input: dict[str, Any], - index: int, - default: str, -) -> str: - objectives = task_input.get("objectives", []) - if index < len(objectives): - return str(objectives[index]) - return default - - -def profile_for(task_type: str) -> dict[str, Any]: - return TASK_TYPE_PROFILES[task_type] - - -def build_design_output(task_input: dict[str, Any]) -> dict[str, Any]: - task_type = str(task_input.get("task_type")) - inputs = subject_inputs(task_input) - profile = profile_for(task_type) - if task_type == "market_research": - return { - "summary": f"Design a concept for {summarize_subject(task_input)}", - "objective": objective_for( - task_input, 0, str(profile["design_default_objective"]) - ), - "subject": inputs, - "focus": profile["design_focus"], - } - if task_type == "product_design": - return { - "summary": ( - f"Shape the product direction for {inputs['subject_name']} with " - f"attention to concept clarity and buildability" - ), - "objective": objective_for( - task_input, 0, str(profile["design_default_objective"]) - ), - "concept_direction": { - "subject": inputs["subject_name"], - "category": inputs["subject_category"], - "hero_attributes": inputs["subject_attributes"][:3], - }, - "focus": profile["design_focus"], - } - return { - "summary": ( - f"Define an improved experience for {inputs['subject_name']} in " - f"{inputs['subject_category']}" - ), - "objective": objective_for( - task_input, 0, str(profile["design_default_objective"]) - ), - "experience_map": { - "subject": inputs["subject_name"], - "primary_friction_signals": inputs["subject_attributes"][:3], - "target_audience": inputs.get("target_audience"), - }, - "focus": profile["design_focus"], - } - - -def build_research_output(task_input: dict[str, Any]) -> dict[str, Any]: - task_type = str(task_input.get("task_type")) - inputs = subject_inputs(task_input) - profile = profile_for(task_type) - if task_type == "market_research": - return { - "summary": f"Research market demand for {inputs['subject_name']}", - "objective": objective_for( - task_input, 1, str(profile["research_default_objective"]) - ), - "market_trends": infer_market_trends(task_input), - "focus": profile["research_focus"], - } - if task_type == "product_design": - return { - "summary": ( - f"Research constraints and adjacent patterns for {inputs['subject_name']}" - ), - "objective": objective_for( - task_input, 1, str(profile["research_default_objective"]) - ), - "design_constraints": infer_design_constraints(task_input), - "focus": profile["research_focus"], - } - return { - "summary": f"Research UX friction patterns for {inputs['subject_name']}", - "objective": objective_for( - task_input, 1, str(profile["research_default_objective"]) - ), - "ux_findings": infer_ux_findings(task_input), - "focus": profile["research_focus"], - } - - -def build_marketing_output(task_input: dict[str, Any]) -> dict[str, Any]: - task_type = str(task_input.get("task_type")) - inputs = subject_inputs(task_input) - profile = profile_for(task_type) - attributes = ", ".join(inputs["subject_attributes"]) - if task_type == "market_research": - return { - "summary": f"Create marketing strategy for {inputs['subject_name']}", - "objective": objective_for( - task_input, 2, str(profile["marketing_default_objective"]) - ), - "positioning": ( - f"{inputs['subject_name']} is positioned as " - f"{with_indefinite_article(inputs['subject_category'].lower())} offer that " - f"combines {attributes} into one concise value story." - ), - "campaign_hooks": [ - f"Lead with {inputs['subject_attributes'][0] if inputs['subject_attributes'] else 'the clearest differentiator'} as the hero differentiator", - "Use research-backed proof points in launch messaging", - "Connect product utility to everyday buyer outcomes", - ], - "focus": profile["marketing_focus"], - } - if task_type == "product_design": - return { - "summary": f"Prepare launch framing for the {inputs['subject_name']} concept", - "objective": objective_for( - task_input, 2, str(profile["marketing_default_objective"]) - ), - "launch_story": ( - f"Present {inputs['subject_name']} as a concept that turns {attributes} " - "into a coherent premium product direction." - ), - "stakeholder_hooks": [ - "Frame the concept as feasible, differentiated, and buyer-relevant", - "Use constraints to explain why the chosen direction is disciplined", - ], - "focus": profile["marketing_focus"], - } - return { - "summary": f"Package the UX rollout plan for {inputs['subject_name']}", - "objective": objective_for( - task_input, 2, str(profile["marketing_default_objective"]) - ), - "rollout_story": ( - f"Position the {inputs['subject_name']} update as a friction-reduction effort " - "that improves confidence and completion rates." - ), - "stakeholder_hooks": [ - "Tie each improvement to a visible user pain point", - "Explain the rollout in terms of reduced friction and clearer trust cues", - ], - "focus": profile["marketing_focus"], - } - - def write_deliverable(output_path: Path, payload: dict[str, Any]) -> Path: output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text( @@ -437,77 +183,59 @@ def write_deliverable(output_path: Path, payload: dict[str, Any]) -> Path: return output_path -def design_task(persona: Persona, context: TaskContext) -> dict[str, object]: - context.update_progress(persona.name, "Designing") - result = { - "persona_name": persona.name, - "role": persona.role, - "skills": persona.skills, - "task_output": build_design_output(context.task_input), - "previous_results": [], - } - context.set_result(persona.name, result) - return result - - -def research_task( +def run_stage( + stage_name: str, persona: Persona, context: TaskContext, - design_persona_name: str, + previous_persona_names: tuple[str, ...] = (), ) -> dict[str, object]: - design_result = context.get_result(design_persona_name) - context.update_progress(persona.name, "Researching") + previous_results: list[dict[str, object]] = [] + for previous_persona_name in previous_persona_names: + previous_result = context.get_result(previous_persona_name) + if previous_result: + previous_results.append(snapshot_result(previous_result)) + + context.update_progress(persona.name, STAGE_PROGRESS[stage_name]) result = { "persona_name": persona.name, "role": persona.role, "skills": persona.skills, - "task_output": build_research_output(context.task_input), - "previous_results": [snapshot_result(design_result)] if design_result else [], + "task_output": build_stage_output( + context.task_type, + stage_name, + context.task_input, + ), + "previous_results": previous_results, } context.set_result(persona.name, result) return result -def marketing_task( - persona: Persona, +def build_final_deliverable( context: TaskContext, design_persona_name: str, research_persona_name: str, -) -> dict[str, object]: - profile = profile_for(context.task_type) - inputs = subject_inputs(context.task_input) + marketing_persona_name: str, +) -> dict[str, Any]: design_result = context.get_result(design_persona_name) research_result = context.get_result(research_persona_name) - context.update_progress(persona.name, "Marketing") - result = { - "persona_name": persona.name, - "role": persona.role, - "skills": persona.skills, - "task_output": build_marketing_output(context.task_input), - "previous_results": [ - snapshot_result(item) for item in [design_result, research_result] if item + marketing_result = context.get_result(marketing_persona_name) + return { + "task_name": context.task_name, + "task_type": context.task_type, + "deliverable_type": deliverable_type_for(context.task_type), + "inputs": context.task_input.get("inputs", {}), + "objectives": context.task_input.get("objectives", []), + "prepared_by": marketing_persona_name, + "persona_sequence": [ + design_persona_name, + research_persona_name, + marketing_persona_name, ], + "design_brief": design_result.get("task_output", {}), + "research_summary": research_result.get("task_output", {}), + "marketing_plan": marketing_result.get("task_output", {}), } - context.set_result(persona.name, result) - context.set_final_deliverable( - { - "task_name": context.task_name, - "task_type": context.task_type, - "deliverable_type": profile["deliverable_type"], - "inputs": inputs, - "objectives": context.task_input.get("objectives", []), - "prepared_by": persona.name, - "persona_sequence": [ - design_persona_name, - research_persona_name, - persona.name, - ], - "design_brief": design_result.get("task_output", {}), - "research_summary": research_result.get("task_output", {}), - "marketing_plan": result["task_output"], - } - ) - return result def workflow(task_input_path: Path, output_path: Path | None = None) -> Path: @@ -530,24 +258,34 @@ def workflow(task_input_path: Path, output_path: Path | None = None) -> Path: print() print("Starting Task:") - print(json.dumps(design_task(design_persona, task_context), indent=2)) - print( - json.dumps( - research_task(research_persona, task_context, design_persona.name), - indent=2, - ) + design_result = run_stage("design", design_persona, task_context) + print(json.dumps(design_result, indent=2)) + + research_result = run_stage( + "research", + research_persona, + task_context, + (design_persona.name,), ) - print( - json.dumps( - marketing_task( - marketing_persona, - task_context, - design_persona.name, - research_persona.name, - ), - indent=2, + print(json.dumps(research_result, indent=2)) + + marketing_result = run_stage( + "marketing", + marketing_persona, + task_context, + (design_persona.name, research_persona.name), + ) + print(json.dumps(marketing_result, indent=2)) + + task_context.set_final_deliverable( + build_final_deliverable( + task_context, + design_persona.name, + research_persona.name, + marketing_persona.name, ) ) + print() task_context.show_summary() print() diff --git a/demos/task_registry.py b/demos/task_registry.py new file mode 100644 index 0000000..070fc49 --- /dev/null +++ b/demos/task_registry.py @@ -0,0 +1,342 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable + +StageOutputBuilder = Callable[[dict[str, Any]], dict[str, Any]] + + +@dataclass(frozen=True) +class TaskTypeDefinition: + task_type: str + deliverable_type: str + handlers: dict[str, StageOutputBuilder] + + +TASK_REGISTRY: dict[str, TaskTypeDefinition] = {} + + +def register_task_type(definition: TaskTypeDefinition) -> None: + TASK_REGISTRY[definition.task_type] = definition + + +def supported_task_types() -> frozenset[str]: + return frozenset(TASK_REGISTRY) + + +def get_task_definition(task_type: str) -> TaskTypeDefinition: + try: + return TASK_REGISTRY[task_type] + except KeyError as exc: + supported = ", ".join(sorted(TASK_REGISTRY)) + raise ValueError( + f"Unsupported `task_type`: {task_type or '(missing)'}. Expected one of: {supported}." + ) from exc + + +def build_stage_output( + task_type: str, + stage_name: str, + task_input: dict[str, Any], +) -> dict[str, Any]: + definition = get_task_definition(task_type) + try: + builder = definition.handlers[stage_name] + except KeyError as exc: + raise ValueError( + f"Task type `{task_type}` does not define a `{stage_name}` handler." + ) from exc + return builder(task_input) + + +def deliverable_type_for(task_type: str) -> str: + return get_task_definition(task_type).deliverable_type + + +def subject_inputs(task_input: dict[str, Any]) -> dict[str, Any]: + return task_input.get("inputs", {}) + + +def summarize_subject(task_input: dict[str, Any]) -> str: + inputs = subject_inputs(task_input) + feature_list = ", ".join(inputs.get("subject_attributes", [])) or "core attributes" + return ( + f"{inputs.get('subject_name', 'Unknown subject')} in " + f"{inputs.get('subject_category', 'Unknown')}" + f" with {feature_list}" + ) + + +def with_indefinite_article(text: str) -> str: + article = "an" if text[:1].lower() in {"a", "e", "i", "o", "u"} else "a" + return f"{article} {text}" + + +def infer_market_trends(task_input: dict[str, Any]) -> list[str]: + inputs = subject_inputs(task_input) + features = {str(feature).lower() for feature in inputs.get("subject_attributes", [])} + trends: list[str] = [] + if "5g" in features: + trends.append("5G adoption remains a purchase driver in premium devices") + if "fast charging" in features: + trends.append("Battery convenience influences upgrade decisions") + if "touchscreen" in features: + trends.append("Large responsive displays remain central to daily usage") + if str(inputs.get("subject_category", "")).lower() == "electronics": + trends.append("Consumers compare feature density with price sensitivity") + return trends or ["General market demand should be validated with recent signals"] + + +def infer_design_constraints(task_input: dict[str, Any]) -> list[str]: + inputs = subject_inputs(task_input) + constraints = inputs.get("constraints", []) + if constraints: + return constraints + return [ + "Balance concept ambition with implementation feasibility", + "Keep differentiation obvious to the target audience", + ] + + +def infer_ux_findings(task_input: dict[str, Any]) -> list[str]: + inputs = subject_inputs(task_input) + attributes = {str(item).lower() for item in inputs.get("subject_attributes", [])} + findings: list[str] = [] + if "form abandonment" in attributes: + findings.append("Long form steps likely interrupt momentum before payment.") + if "payment trust" in attributes: + findings.append("Trust cues near payment decisions need to be more visible.") + if "mobile checkout" in attributes: + findings.append("Mobile users need fewer taps and clearer progress markers.") + return findings or [ + "Review friction around the primary user path and prioritize the highest drop-off points." + ] + + +def objective_for( + task_input: dict[str, Any], + index: int, + default: str, +) -> str: + objectives = task_input.get("objectives", []) + if index < len(objectives): + return str(objectives[index]) + return default + + +def build_market_research_design_output(task_input: dict[str, Any]) -> dict[str, Any]: + return { + "summary": f"Design a concept for {summarize_subject(task_input)}", + "objective": objective_for(task_input, 0, "Design a product concept"), + "subject": subject_inputs(task_input), + "focus": [ + "clarify the value proposition", + "connect concept choices to buyer expectations", + "prepare a market-aware design brief", + ], + } + + +def build_market_research_research_output(task_input: dict[str, Any]) -> dict[str, Any]: + inputs = subject_inputs(task_input) + return { + "summary": f"Research market demand for {inputs['subject_name']}", + "objective": objective_for(task_input, 1, "Research market trends"), + "market_trends": infer_market_trends(task_input), + "focus": [ + "collect supporting market signals", + "distill concise findings", + "translate evidence into technical context", + ], + } + + +def build_market_research_marketing_output(task_input: dict[str, Any]) -> dict[str, Any]: + inputs = subject_inputs(task_input) + attributes = ", ".join(inputs["subject_attributes"]) + hero_attribute = ( + inputs["subject_attributes"][0] + if inputs["subject_attributes"] + else "the clearest differentiator" + ) + return { + "summary": f"Create marketing strategy for {inputs['subject_name']}", + "objective": objective_for(task_input, 2, "Create marketing strategy"), + "positioning": ( + f"{inputs['subject_name']} is positioned as " + f"{with_indefinite_article(inputs['subject_category'].lower())} offer that " + f"combines {attributes} into one concise value story." + ), + "campaign_hooks": [ + f"Lead with {hero_attribute} as the hero differentiator", + "Use research-backed proof points in launch messaging", + "Connect product utility to everyday buyer outcomes", + ], + "focus": [ + "turn research into positioning", + "shape a campaign-ready narrative", + "package the final deliverable", + ], + } + + +def build_product_design_design_output(task_input: dict[str, Any]) -> dict[str, Any]: + inputs = subject_inputs(task_input) + return { + "summary": ( + f"Shape the product direction for {inputs['subject_name']} with " + "attention to concept clarity and buildability" + ), + "objective": objective_for(task_input, 0, "Shape the product concept"), + "concept_direction": { + "subject": inputs["subject_name"], + "category": inputs["subject_category"], + "hero_attributes": inputs["subject_attributes"][:3], + }, + "focus": [ + "define the concept direction", + "prioritize user-facing differentiators", + "keep the concept buildable under constraints", + ], + } + + +def build_product_design_research_output(task_input: dict[str, Any]) -> dict[str, Any]: + inputs = subject_inputs(task_input) + return { + "summary": f"Research constraints and adjacent patterns for {inputs['subject_name']}", + "objective": objective_for( + task_input, + 1, + "Review design constraints and competitive cues", + ), + "design_constraints": infer_design_constraints(task_input), + "focus": [ + "compare adjacent product patterns", + "identify feasibility and usability constraints", + "surface practical tradeoffs for the concept", + ], + } + + +def build_product_design_marketing_output(task_input: dict[str, Any]) -> dict[str, Any]: + inputs = subject_inputs(task_input) + attributes = ", ".join(inputs["subject_attributes"]) + return { + "summary": f"Prepare launch framing for the {inputs['subject_name']} concept", + "objective": objective_for( + task_input, + 2, + "Prepare launch messaging for the concept", + ), + "launch_story": ( + f"Present {inputs['subject_name']} as a concept that turns {attributes} " + "into a coherent premium product direction." + ), + "stakeholder_hooks": [ + "Frame the concept as feasible, differentiated, and buyer-relevant", + "Use constraints to explain why the chosen direction is disciplined", + ], + "focus": [ + "translate the concept into launch language", + "align story with buyer expectations", + "package the concept as a reviewable brief", + ], + } + + +def build_ux_review_design_output(task_input: dict[str, Any]) -> dict[str, Any]: + inputs = subject_inputs(task_input) + return { + "summary": ( + f"Define an improved experience for {inputs['subject_name']} in " + f"{inputs['subject_category']}" + ), + "objective": objective_for(task_input, 0, "Define a better checkout experience"), + "experience_map": { + "subject": inputs["subject_name"], + "primary_friction_signals": inputs["subject_attributes"][:3], + "target_audience": inputs.get("target_audience"), + }, + "focus": [ + "map the intended user journey", + "reduce points of hesitation", + "prepare a concise improvement concept", + ], + } + + +def build_ux_review_research_output(task_input: dict[str, Any]) -> dict[str, Any]: + inputs = subject_inputs(task_input) + return { + "summary": f"Research UX friction patterns for {inputs['subject_name']}", + "objective": objective_for(task_input, 1, "Identify friction and usage patterns"), + "ux_findings": infer_ux_findings(task_input), + "focus": [ + "identify friction signals", + "connect issues to user behavior patterns", + "prioritize what should change first", + ], + } + + +def build_ux_review_marketing_output(task_input: dict[str, Any]) -> dict[str, Any]: + inputs = subject_inputs(task_input) + return { + "summary": f"Package the UX rollout plan for {inputs['subject_name']}", + "objective": objective_for( + task_input, + 2, + "Package the UX improvement plan for rollout", + ), + "rollout_story": ( + f"Position the {inputs['subject_name']} update as a friction-reduction effort " + "that improves confidence and completion rates." + ), + "stakeholder_hooks": [ + "Tie each improvement to a visible user pain point", + "Explain the rollout in terms of reduced friction and clearer trust cues", + ], + "focus": [ + "frame the UX work for stakeholders", + "turn findings into an adoption story", + "package a rollout-ready improvement plan", + ], + } + + +register_task_type( + TaskTypeDefinition( + task_type="market_research", + deliverable_type="market_strategy_report", + handlers={ + "design": build_market_research_design_output, + "research": build_market_research_research_output, + "marketing": build_market_research_marketing_output, + }, + ) +) + +register_task_type( + TaskTypeDefinition( + task_type="product_design", + deliverable_type="product_concept_brief", + handlers={ + "design": build_product_design_design_output, + "research": build_product_design_research_output, + "marketing": build_product_design_marketing_output, + }, + ) +) + +register_task_type( + TaskTypeDefinition( + task_type="ux_review", + deliverable_type="ux_improvement_plan", + handlers={ + "design": build_ux_review_design_output, + "research": build_ux_review_research_output, + "marketing": build_ux_review_marketing_output, + }, + ) +)