diff --git a/src/scribae/feedback.py b/src/scribae/feedback.py new file mode 100644 index 0000000..fcb7eee --- /dev/null +++ b/src/scribae/feedback.py @@ -0,0 +1,680 @@ +from __future__ import annotations + +import asyncio +import json +import re +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Literal, cast + +import frontmatter +from pydantic import BaseModel, ConfigDict, ValidationError, field_validator +from pydantic_ai import Agent, NativeOutput, UnexpectedModelBehavior +from pydantic_ai.settings import ModelSettings + +from .brief import SeoBrief +from .io_utils import NoteDetails, load_note +from .language import LanguageMismatchError, LanguageResolutionError, ensure_language_output, resolve_output_language +from .llm import LLM_OUTPUT_RETRIES, LLM_TIMEOUT_SECONDS, OpenAISettings, make_model +from .project import ProjectConfig +from .prompts.feedback import FEEDBACK_SYSTEM_PROMPT, FeedbackPromptBundle, build_feedback_prompt_bundle + +Reporter = Callable[[str], None] | None + + +class FeedbackError(Exception): + """Base class for feedback command failures.""" + + exit_code = 1 + + def __init__(self, message: str, *, exit_code: int | None = None) -> None: + super().__init__(message) + if exit_code is not None: + self.exit_code = exit_code + + +class FeedbackValidationError(FeedbackError): + exit_code = 2 + + +class FeedbackFileError(FeedbackError): + exit_code = 3 + + +class FeedbackLLMError(FeedbackError): + exit_code = 4 + + +class FeedbackBriefError(FeedbackError): + exit_code = 5 + + +class FeedbackLocation(BaseModel): + """Approximate location of a finding in the draft.""" + + model_config = ConfigDict(extra="forbid") + + heading: str | None = None + paragraph_index: int | None = None + + @field_validator("heading", mode="before") + @classmethod + def _strip_heading(cls, value: str | None) -> str | None: + if value is None: + return None + return str(value).strip() or None + + @field_validator("paragraph_index", mode="before") + @classmethod + def _coerce_index(cls, value: Any) -> int | None: + if value is None: + return None + return int(value) + + +class FeedbackFinding(BaseModel): + """A single issue or observation in the review.""" + + model_config = ConfigDict(extra="forbid") + + severity: Literal["low", "medium", "high"] + category: Literal["seo", "structure", "clarity", "style", "evidence", "other"] + message: str + location: FeedbackLocation | None = None + + @field_validator("message", mode="before") + @classmethod + def _strip_message(cls, value: str) -> str: + return str(value).strip() + + +class FeedbackSummary(BaseModel): + model_config = ConfigDict(extra="forbid") + + issues: list[str] + strengths: list[str] + + @field_validator("issues", "strengths", mode="before") + @classmethod + def _normalize_list(cls, value: Any) -> list[str]: + if isinstance(value, str): + value = [value] + if not isinstance(value, list): + raise TypeError("value must be a list") + return [str(item).strip() for item in value if str(item).strip()] + + +class BriefAlignment(BaseModel): + model_config = ConfigDict(extra="forbid") + + intent: str + outline_covered: list[str] + outline_missing: list[str] + keywords_covered: list[str] + keywords_missing: list[str] + faq_covered: list[str] + faq_missing: list[str] + + @field_validator( + "intent", + mode="before", + ) + @classmethod + def _strip_intent(cls, value: str) -> str: + return str(value).strip() + + @field_validator( + "outline_covered", + "outline_missing", + "keywords_covered", + "keywords_missing", + "faq_covered", + "faq_missing", + mode="before", + ) + @classmethod + def _normalize_list(cls, value: Any) -> list[str]: + if isinstance(value, str): + value = [value] + if not isinstance(value, list): + raise TypeError("value must be a list") + return [str(item).strip() for item in value if str(item).strip()] + + +class SectionNote(BaseModel): + model_config = ConfigDict(extra="forbid") + + heading: str + notes: list[str] + + @field_validator("heading", mode="before") + @classmethod + def _strip_heading(cls, value: str) -> str: + return str(value).strip() + + @field_validator("notes", mode="before") + @classmethod + def _normalize_list(cls, value: Any) -> list[str]: + if isinstance(value, str): + value = [value] + if not isinstance(value, list): + raise TypeError("value must be a list") + return [str(item).strip() for item in value if str(item).strip()] + + +class FeedbackReport(BaseModel): + """Structured report returned by the feedback agent.""" + + model_config = ConfigDict(extra="forbid") + + summary: FeedbackSummary + brief_alignment: BriefAlignment + section_notes: list[SectionNote] + evidence_gaps: list[str] + findings: list[FeedbackFinding] + checklist: list[str] + + @field_validator("evidence_gaps", "checklist", mode="before") + @classmethod + def _normalize_list(cls, value: Any) -> list[str]: + if isinstance(value, str): + value = [value] + if not isinstance(value, list): + raise TypeError("value must be a list") + return [str(item).strip() for item in value if str(item).strip()] + + +class FeedbackFormat(str): + MARKDOWN = "md" + JSON = "json" + BOTH = "both" + + @classmethod + def from_raw(cls, value: str) -> FeedbackFormat: + lowered = value.lower().strip() + if lowered not in {cls.MARKDOWN, cls.JSON, cls.BOTH}: + raise FeedbackValidationError("--format must be md, json, or both.") + return cls(lowered) + + +class FeedbackFocus(str): + SEO = "seo" + STRUCTURE = "structure" + CLARITY = "clarity" + STYLE = "style" + EVIDENCE = "evidence" + + @classmethod + def from_raw(cls, value: str) -> FeedbackFocus: + lowered = value.lower().strip() + allowed = {cls.SEO, cls.STRUCTURE, cls.CLARITY, cls.STYLE, cls.EVIDENCE} + if lowered not in allowed: + raise FeedbackValidationError("--focus must be seo, structure, clarity, style, or evidence.") + return cls(lowered) + + +@dataclass(frozen=True) +class BodySection: + heading: str + content: str + index: int + + +@dataclass(frozen=True) +class BodyDocument: + path: Path + content: str + excerpt: str + frontmatter: dict[str, Any] + truncated: bool + + +@dataclass(frozen=True) +class FeedbackContext: + body: BodyDocument + brief: SeoBrief + project: ProjectConfig + note: NoteDetails | None + focus: str | None + language: str + selected_outline: list[str] + selected_sections: list[BodySection] + section_range: tuple[int, int] | None + + +PromptBundle = FeedbackPromptBundle +SYSTEM_PROMPT = FEEDBACK_SYSTEM_PROMPT + + +def prepare_context( + *, + body_path: Path, + brief_path: Path, + project: ProjectConfig, + note_path: Path | None = None, + language: str | None = None, + focus: str | None = None, + section_range: tuple[int, int] | None = None, + max_body_chars: int = 12000, + max_note_chars: int = 6000, + language_detector: Callable[[str], str] | None = None, + reporter: Reporter = None, +) -> FeedbackContext: + """Load inputs and prepare the feedback context.""" + if max_body_chars <= 0 or max_note_chars <= 0: + raise FeedbackValidationError("Max chars must be greater than zero.") + + body = _load_body(body_path, max_chars=max_body_chars) + brief = _load_brief(brief_path) + note = _load_note(note_path, max_chars=max_note_chars) if note_path else None + + _report(reporter, f"Loaded draft '{body.path.name}' and brief '{brief.title}'.") + if note is not None: + _report(reporter, f"Loaded source note '{note.title}'.") + + try: + language_resolution = resolve_output_language( + flag_language=language, + project_language=project.get("language"), + metadata=body.frontmatter, + text=body.content, + language_detector=language_detector, + ) + except LanguageResolutionError as exc: + raise FeedbackValidationError(str(exc)) from exc + + _report( + reporter, + f"Resolved output language: {language_resolution.language} (source: {language_resolution.source})", + ) + + selected_outline = _select_outline(brief, section_range=section_range) + sections = _split_body_sections(body.content) + selected_sections = _select_body_sections(sections, section_range=section_range) + + return FeedbackContext( + body=body, + brief=brief, + project=project, + note=note, + focus=focus, + language=language_resolution.language, + selected_outline=selected_outline, + selected_sections=selected_sections, + section_range=section_range, + ) + + +def build_prompt_bundle(context: FeedbackContext) -> FeedbackPromptBundle: + return build_feedback_prompt_bundle(_prompt_context(context)) + + +def render_dry_run_prompt(context: FeedbackContext) -> str: + prompts = build_prompt_bundle(context) + return prompts.user_prompt + + +def generate_feedback_report( + context: FeedbackContext, + *, + model_name: str, + temperature: float, + reporter: Reporter = None, + agent: Agent[None, FeedbackReport] | None = None, + prompts: PromptBundle | None = None, + timeout_seconds: float = LLM_TIMEOUT_SECONDS, + language_detector: Callable[[str], str] | None = None, +) -> FeedbackReport: + """Generate the structured feedback report via the LLM.""" + prompts = prompts or build_prompt_bundle(context) + + resolved_settings = OpenAISettings.from_env() + llm_agent: Agent[None, FeedbackReport] = ( + agent if agent is not None else _create_agent(model_name, temperature) + ) + + _report(reporter, f"Calling model '{model_name}' via {resolved_settings.base_url}") + + try: + report = cast( + FeedbackReport, + ensure_language_output( + prompt=prompts.user_prompt, + expected_language=context.language, + invoke=lambda prompt: _invoke_agent(llm_agent, prompt, timeout_seconds=timeout_seconds), + extract_text=_feedback_language_text, + reporter=reporter, + language_detector=language_detector, + ), + ) + except UnexpectedModelBehavior as exc: + raise FeedbackValidationError(f"Model returned unexpected output: {exc}") from exc + except LanguageMismatchError as exc: + raise FeedbackValidationError(str(exc)) from exc + except LanguageResolutionError as exc: + raise FeedbackValidationError(str(exc)) from exc + except TimeoutError as exc: + raise FeedbackLLMError(f"LLM request timed out after {int(timeout_seconds)} seconds.") from exc + except FeedbackError: + raise + except Exception as exc: # pragma: no cover - surfaced to CLI + raise FeedbackLLMError(f"LLM request failed: {exc}") from exc + + return report + + +def render_json(report: FeedbackReport) -> str: + return json.dumps(report.model_dump(), indent=2, ensure_ascii=False) + + +def render_markdown(report: FeedbackReport) -> str: + sections: list[str] = ["# Feedback Report", "", "## Summary", "", "### Top issues"] + sections.extend(_render_list(report.summary.issues)) + sections.append("") + sections.append("### Strengths") + sections.extend(_render_list(report.summary.strengths)) + sections.append("") + sections.append("## Brief alignment") + sections.append("") + sections.append(f"- Intent: {report.brief_alignment.intent}") + sections.append(f"- Outline covered: {', '.join(report.brief_alignment.outline_covered) or 'None noted'}") + sections.append(f"- Outline missing: {', '.join(report.brief_alignment.outline_missing) or 'None noted'}") + sections.append(f"- Keywords covered: {', '.join(report.brief_alignment.keywords_covered) or 'None noted'}") + sections.append(f"- Keywords missing: {', '.join(report.brief_alignment.keywords_missing) or 'None noted'}") + sections.append(f"- FAQ covered: {', '.join(report.brief_alignment.faq_covered) or 'None noted'}") + sections.append(f"- FAQ missing: {', '.join(report.brief_alignment.faq_missing) or 'None noted'}") + sections.append("") + sections.append("## Section notes") + sections.append("") + if report.section_notes: + for item in report.section_notes: + sections.append(f"### {item.heading}") + sections.extend(_render_list(item.notes)) + sections.append("") + else: + sections.extend(_render_list([])) + sections.append("") + + sections.append("## Evidence gaps") + sections.append("") + sections.extend(_render_list(report.evidence_gaps)) + sections.append("") + + sections.append("## Findings") + sections.append("") + if report.findings: + for finding in report.findings: + location = _format_location(finding.location) + sections.append( + f"- **{finding.severity.upper()}** [{finding.category}] {finding.message}{location}" + ) + else: + sections.extend(_render_list([])) + sections.append("") + + sections.append("## Checklist") + sections.append("") + if report.checklist: + sections.extend([f"- [ ] {item}" for item in report.checklist]) + else: + sections.extend(_render_list([])) + sections.append("") + + return "\n".join(sections).rstrip() + "\n" + + +def save_prompt_artifacts( + prompts: PromptBundle, + *, + destination: Path, + response: FeedbackReport | None = None, +) -> tuple[Path, Path | None]: + destination.mkdir(parents=True, exist_ok=True) + prompt_path = destination / "feedback.prompt.txt" + response_path: Path | None = destination / "feedback.response.json" if response is not None else None + + prompt_payload = f"SYSTEM PROMPT:\n{prompts.system_prompt}\n\nUSER PROMPT:\n{prompts.user_prompt}\n" + prompt_path.write_text(prompt_payload, encoding="utf-8") + + if response is not None and response_path is not None: + response_path.write_text(render_json(response) + "\n", encoding="utf-8") + + return prompt_path, response_path + + +def parse_section_range(value: str) -> tuple[int, int]: + match = re.fullmatch(r"(\d+)\.\.(\d+)", value.strip()) + if not match: + raise FeedbackValidationError("--section must use the format N..M (e.g., 2..4).") + start, end = int(match.group(1)), int(match.group(2)) + if start <= 0 or end <= 0: + raise FeedbackValidationError("Section numbers must be positive.") + if start > end: + raise FeedbackValidationError("Section range start must be <= end.") + return start, end + + +def _prompt_context(context: FeedbackContext) -> _FeedbackPromptContext: + selected_sections = [ + {"heading": section.heading, "content": section.content} for section in context.selected_sections + ] + return _FeedbackPromptContext( + brief=context.brief, + body=context.body, + note_excerpt=context.note.body if context.note else None, + project=context.project, + language=context.language, + focus=context.focus, + selected_outline=context.selected_outline, + selected_sections=selected_sections, + ) + + +@dataclass(frozen=True) +class _FeedbackPromptContext: + brief: SeoBrief + body: BodyDocument + note_excerpt: str | None + project: ProjectConfig + language: str + focus: str | None + selected_outline: list[str] + selected_sections: list[dict[str, str]] + + +def _create_agent(model_name: str, temperature: float) -> Agent[None, FeedbackReport]: + model_settings = ModelSettings(temperature=temperature) + model = make_model(model_name, model_settings=model_settings) + return Agent[None, FeedbackReport]( + model=model, + output_type=NativeOutput(FeedbackReport, name="FeedbackReport", strict=True), + instructions=SYSTEM_PROMPT, + output_retries=LLM_OUTPUT_RETRIES, + ) + + +def _invoke_agent(agent: Agent[None, FeedbackReport], prompt: str, *, timeout_seconds: float) -> FeedbackReport: + async def _call() -> FeedbackReport: + run = await agent.run(prompt) + output = getattr(run, "output", None) + if isinstance(output, FeedbackReport): + return output + if isinstance(output, BaseModel): + return FeedbackReport.model_validate(output.model_dump()) + if isinstance(output, dict): + return FeedbackReport.model_validate(output) + raise TypeError("LLM output is not a FeedbackReport instance") + + return asyncio.run(asyncio.wait_for(_call(), timeout_seconds)) + + +def _feedback_language_text(report: FeedbackReport) -> str: + issue_text = " ".join(report.summary.issues) + strength_text = " ".join(report.summary.strengths) + findings = " ".join([finding.message for finding in report.findings]) + checklist = " ".join(report.checklist) + section_notes = " ".join([" ".join(item.notes) for item in report.section_notes]) + return "\n".join([issue_text, strength_text, findings, checklist, section_notes]).strip() + + +def _load_body(body_path: Path, *, max_chars: int) -> BodyDocument: + try: + post = frontmatter.load(body_path) + except FileNotFoundError as exc: + raise FeedbackFileError(f"Draft file not found: {body_path}") from exc + except OSError as exc: # pragma: no cover - surfaced by CLI + raise FeedbackFileError(f"Unable to read draft: {exc}") from exc + except Exception as exc: # pragma: no cover - parsing errors + raise FeedbackFileError(f"Unable to parse draft {body_path}: {exc}") from exc + + metadata = dict(post.metadata or {}) + content = post.content.strip() + excerpt, truncated = _truncate(content, max_chars) + return BodyDocument( + path=body_path, + content=excerpt, + excerpt=excerpt, + frontmatter=metadata, + truncated=truncated, + ) + + +def _load_brief(path: Path) -> SeoBrief: + try: + text = path.read_text(encoding="utf-8") + except FileNotFoundError as exc: + raise FeedbackBriefError(f"Brief JSON not found: {path}") from exc + except OSError as exc: # pragma: no cover - surfaced by CLI + raise FeedbackBriefError(f"Unable to read brief: {exc}") from exc + + try: + payload = json.loads(text) + except json.JSONDecodeError as exc: + raise FeedbackBriefError(f"Brief file is not valid JSON: {exc}") from exc + + try: + return SeoBrief.model_validate(payload) + except ValidationError as exc: + raise FeedbackBriefError(f"Brief JSON failed validation: {exc}") from exc + + +def _load_note(note_path: Path, *, max_chars: int) -> NoteDetails: + try: + return load_note(note_path, max_chars=max_chars) + except FileNotFoundError as exc: + raise FeedbackFileError(f"Note file not found: {note_path}") from exc + except ValueError as exc: + raise FeedbackFileError(str(exc)) from exc + except OSError as exc: # pragma: no cover - surfaced by CLI + raise FeedbackFileError(f"Unable to read note: {exc}") from exc + + +def _select_outline(brief: SeoBrief, *, section_range: tuple[int, int] | None) -> list[str]: + if not brief.outline: + raise FeedbackValidationError("Brief outline is empty.") + + total = len(brief.outline) + start, end = (1, total) + if section_range is not None: + start, end = section_range + if not (1 <= start <= total and 1 <= end <= total and start <= end): + raise FeedbackValidationError(f"Section range {start}..{end} is invalid for {total} outline items.") + + selected = [brief.outline[idx - 1].strip() for idx in range(start, end + 1) if brief.outline[idx - 1].strip()] + if not selected: + raise FeedbackValidationError("No outline sections selected.") + return selected + + +def _split_body_sections(body: str) -> list[BodySection]: + heading_pattern = re.compile(r"^(#{1,3})\s+(.*)$") + sections: list[BodySection] = [] + current_heading: str | None = None + current_lines: list[str] = [] + + def _flush() -> None: + if current_heading is None and not current_lines: + return + heading = current_heading or "Body" + content = "\n".join(current_lines).strip() + sections.append(BodySection(heading=heading, content=content, index=len(sections) + 1)) + + for line in body.splitlines(): + match = heading_pattern.match(line.strip()) + if match: + _flush() + current_heading = match.group(2).strip() or "Untitled" + current_lines = [] + continue + current_lines.append(line) + + _flush() + + if not sections: + return [BodySection(heading="Body", content=body.strip(), index=1)] + return sections + + +def _select_body_sections( + sections: Sequence[BodySection], + *, + section_range: tuple[int, int] | None, +) -> list[BodySection]: + if section_range is None: + return list(sections) + + start, end = section_range + selected = [section for section in sections if start <= section.index <= end] + return selected or list(sections) + + +def _render_list(items: Sequence[str]) -> list[str]: + if not items: + return ["- None noted."] + return [f"- {item}" for item in items] + + +def _format_location(location: FeedbackLocation | None) -> str: + if location is None: + return "" + heading = location.heading + paragraph = location.paragraph_index + details: list[str] = [] + if heading: + details.append(f"heading: {heading}") + if paragraph is not None: + details.append(f"paragraph: {paragraph}") + return f" ({'; '.join(details)})" if details else "" + + +def _truncate(value: str, max_chars: int) -> tuple[str, bool]: + if len(value) <= max_chars: + return value, False + return value[: max_chars - 1].rstrip() + " …", True + + +def _report(reporter: Reporter, message: str) -> None: + if reporter: + reporter(message) + + +__all__ = [ + "FeedbackReport", + "FeedbackFormat", + "FeedbackFocus", + "FeedbackContext", + "FeedbackError", + "FeedbackValidationError", + "FeedbackFileError", + "FeedbackLLMError", + "FeedbackBriefError", + "prepare_context", + "build_prompt_bundle", + "render_dry_run_prompt", + "generate_feedback_report", + "render_json", + "render_markdown", + "save_prompt_artifacts", + "parse_section_range", +] diff --git a/src/scribae/feedback_cli.py b/src/scribae/feedback_cli.py new file mode 100644 index 0000000..6210605 --- /dev/null +++ b/src/scribae/feedback_cli.py @@ -0,0 +1,278 @@ +from __future__ import annotations + +from pathlib import Path + +import typer + +from .feedback import ( + FeedbackBriefError, + FeedbackError, + FeedbackFileError, + FeedbackFocus, + FeedbackFormat, + FeedbackReport, + FeedbackValidationError, + build_prompt_bundle, + generate_feedback_report, + parse_section_range, + prepare_context, + render_dry_run_prompt, + render_json, + render_markdown, + save_prompt_artifacts, +) +from .llm import DEFAULT_MODEL_NAME +from .project import load_default_project, load_project + + +def feedback_command( + body: Path = typer.Option( # noqa: B008 + ..., + "--body", + "-b", + resolve_path=True, + help="Path to the Markdown draft to review.", + ), + brief: Path = typer.Option( # noqa: B008 + ..., + "--brief", + help="Path to the SeoBrief JSON output from `scribae brief`.", + ), + note: Path | None = typer.Option( # noqa: B008 + None, + "--note", + "-n", + resolve_path=True, + help="Optional source note for grounding and fact checks.", + ), + section: str | None = typer.Option( # noqa: B008 + None, + "--section", + help="Limit review to outline section range N..M (1-indexed).", + ), + focus: str | None = typer.Option( # noqa: B008 + None, + "--focus", + help="Narrow the review scope: seo|structure|clarity|style|evidence.", + ), + output_format: str = typer.Option( # noqa: B008 + FeedbackFormat.MARKDOWN, + "--format", + "-f", + help="Output format: md|json|both.", + case_sensitive=False, + ), + out: Path | None = typer.Option( # noqa: B008 + None, + "--out", + "-o", + resolve_path=True, + help="Write output to this file (stdout if omitted).", + ), + out_dir: Path | None = typer.Option( # noqa: B008 + None, + "--out-dir", + resolve_path=True, + help="Directory to write outputs when using --format both.", + ), + project: str | None = typer.Option( # noqa: B008 + None, + "--project", + "-p", + help="Project name (loads .yml/.yaml from current directory) or path to a project file.", + ), + language: str | None = typer.Option( # noqa: B008 + None, + "--language", + "-l", + help="Language code for the feedback report (overrides project config).", + ), + model: str = typer.Option( # noqa: B008 + DEFAULT_MODEL_NAME, + "--model", + "-m", + help="Model name to request via OpenAI-compatible API.", + ), + temperature: float = typer.Option( # noqa: B008 + 0.2, + "--temperature", + min=0.0, + max=2.0, + help="Temperature for the LLM request.", + ), + dry_run: bool = typer.Option( # noqa: B008 + False, + "--dry-run", + help="Print the generated prompt and skip the LLM call.", + ), + save_prompt: Path | None = typer.Option( # noqa: B008 + None, + "--save-prompt", + file_okay=False, + dir_okay=True, + exists=False, + resolve_path=True, + help="Directory for saving prompt/response artifacts.", + ), + verbose: bool = typer.Option( # noqa: B008 + False, + "--verbose", + "-v", + help="Print progress information to stderr.", + ), +) -> None: + """Review a draft against an SEO brief without rewriting the draft. + + Examples: + scribae feedback --body draft.md --brief brief.json + scribae feedback --body draft.md --brief brief.json --format json --out feedback.json + scribae feedback --body draft.md --brief brief.json --section 1..3 --focus seo + """ + reporter = (lambda msg: typer.secho(msg, err=True)) if verbose else None + + try: + fmt = FeedbackFormat.from_raw(output_format) + except FeedbackValidationError as exc: + typer.secho(str(exc), err=True, fg=typer.colors.RED) + raise typer.Exit(exc.exit_code) from exc + + if dry_run and (out is not None or out_dir is not None or save_prompt is not None): + raise typer.BadParameter("--dry-run cannot be combined with output options.", param_hint="--dry-run") + + if fmt == FeedbackFormat.BOTH: + if out is not None and out_dir is not None: + raise typer.BadParameter("Use --out or --out-dir, not both, with --format both.") + if out is None and out_dir is None: + raise typer.BadParameter("--format both requires --out or --out-dir.") + + if project: + try: + project_config = load_project(project) + except (FileNotFoundError, ValueError, OSError) as exc: + typer.secho(str(exc), err=True, fg=typer.colors.RED) + raise typer.Exit(5) from exc + else: + try: + project_config, project_source = load_default_project() + except (FileNotFoundError, ValueError, OSError) as exc: + typer.secho(str(exc), err=True, fg=typer.colors.RED) + raise typer.Exit(5) from exc + if not project_source: + typer.secho( + "No project provided; using default context (language=en, tone=neutral).", + err=True, + fg=typer.colors.YELLOW, + ) + + section_range = None + if section: + try: + section_range = parse_section_range(section) + except FeedbackValidationError as exc: + typer.secho(str(exc), err=True, fg=typer.colors.RED) + raise typer.Exit(exc.exit_code) from exc + + focus_value = None + if focus: + try: + focus_value = str(FeedbackFocus.from_raw(focus)) + except FeedbackValidationError as exc: + typer.secho(str(exc), err=True, fg=typer.colors.RED) + raise typer.Exit(exc.exit_code) from exc + + try: + context = prepare_context( + body_path=body.expanduser(), + brief_path=brief.expanduser(), + note_path=note.expanduser() if note else None, + project=project_config, + language=language, + focus=focus_value, + section_range=section_range, + reporter=reporter, + ) + except (FeedbackBriefError, FeedbackFileError, FeedbackValidationError, FeedbackError) as exc: + typer.secho(str(exc), err=True, fg=typer.colors.RED) + raise typer.Exit(exc.exit_code) from exc + + prompts = build_prompt_bundle(context) + + if dry_run: + typer.echo(render_dry_run_prompt(context)) + return + + try: + report = generate_feedback_report( + context, + model_name=model, + temperature=temperature, + reporter=reporter, + prompts=prompts, + ) + except KeyboardInterrupt: + typer.secho("Cancelled by user.", err=True, fg=typer.colors.YELLOW) + raise typer.Exit(130) from None + except FeedbackError as exc: + typer.secho(str(exc), err=True, fg=typer.colors.RED) + raise typer.Exit(exc.exit_code) from exc + + if save_prompt is not None: + try: + save_prompt_artifacts(prompts, destination=save_prompt, response=report) + except OSError as exc: + typer.secho(f"Unable to save prompt artifacts: {exc}", err=True, fg=typer.colors.RED) + raise typer.Exit(3) from exc + + _write_outputs(report, fmt=fmt, out=out, out_dir=out_dir) + + +def _write_outputs( + report: FeedbackReport, + *, + fmt: FeedbackFormat, + out: Path | None, + out_dir: Path | None, +) -> None: + if fmt == FeedbackFormat.JSON: + payload = render_json(report) + _write_single_output(payload, out, label="feedback JSON") + return + + if fmt == FeedbackFormat.MARKDOWN: + payload = render_markdown(report) + _write_single_output(payload, out, label="feedback Markdown") + return + + md_payload = render_markdown(report) + json_payload = render_json(report) + + if out_dir is not None: + out_dir.mkdir(parents=True, exist_ok=True) + md_path = out_dir / "feedback.md" + json_path = out_dir / "feedback.json" + else: + assert out is not None + if out.suffix.lower() == ".json": + json_path = out + md_path = out.with_suffix(".md") + else: + md_path = out + json_path = out.with_suffix(out.suffix + ".json") + + md_path.parent.mkdir(parents=True, exist_ok=True) + md_path.write_text(md_payload, encoding="utf-8") + json_path.write_text(json_payload + "\n", encoding="utf-8") + typer.echo(f"Wrote feedback Markdown to {md_path}") + typer.echo(f"Wrote feedback JSON to {json_path}") + + +def _write_single_output(payload: str, out: Path | None, *, label: str) -> None: + if out is None: + typer.echo(payload, nl=False) + return + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text(payload + ("" if payload.endswith("\n") else "\n"), encoding="utf-8") + typer.echo(f"Wrote {label} to {out}") + + +__all__ = ["feedback_command"] diff --git a/src/scribae/main.py b/src/scribae/main.py index 80a75ee..7b42e03 100644 --- a/src/scribae/main.py +++ b/src/scribae/main.py @@ -3,6 +3,7 @@ import typer from .brief_cli import brief_command +from .feedback_cli import feedback_command from .idea_cli import idea_command from .meta_cli import meta_command from .translate_cli import translate_command @@ -30,6 +31,10 @@ def app_callback() -> None: help="Generate a validated SEO brief (keywords, outline, FAQ, metadata) from a note.", )(brief_command) app.command("write", help="Draft an article from a note + SeoBrief JSON.")(write_command) +app.command( + "feedback", + help="Review a draft against a brief to surface improvements without rewriting.", +)(feedback_command) app.command("meta", help="Create publication metadata/frontmatter for a finished draft.")(meta_command) app.command("translate", help="Translate Markdown while preserving formatting (MT + post-edit).")(translate_command) app.command("version", help="Print the Scribae version.")(version_command) diff --git a/src/scribae/prompts/__init__.py b/src/scribae/prompts/__init__.py index 776329a..fe199f1 100644 --- a/src/scribae/prompts/__init__.py +++ b/src/scribae/prompts/__init__.py @@ -5,4 +5,5 @@ "write", "idea", "meta", + "feedback", ] diff --git a/src/scribae/prompts/feedback.py b/src/scribae/prompts/feedback.py new file mode 100644 index 0000000..30c7948 --- /dev/null +++ b/src/scribae/prompts/feedback.py @@ -0,0 +1,172 @@ +from __future__ import annotations + +import json +import textwrap +from dataclasses import dataclass +from typing import Protocol + +from scribae.brief import SeoBrief +from scribae.project import ProjectConfig + + +class FeedbackPromptBody(Protocol): + @property + def excerpt(self) -> str: ... + + +class FeedbackPromptContext(Protocol): + @property + def brief(self) -> SeoBrief: ... + + @property + def body(self) -> FeedbackPromptBody: ... + + @property + def note_excerpt(self) -> str | None: ... + + @property + def project(self) -> ProjectConfig: ... + + @property + def language(self) -> str: ... + + @property + def focus(self) -> str | None: ... + + @property + def selected_outline(self) -> list[str]: ... + + @property + def selected_sections(self) -> list[dict[str, str]]: ... + + +@dataclass(frozen=True) +class FeedbackPromptBundle: + system_prompt: str + user_prompt: str + + +FEEDBACK_SYSTEM_PROMPT = textwrap.dedent( + """\ + You are a meticulous editorial reviewer for SEO drafts. + + Your task is to review a draft against its validated SEO brief (and optional source note) + and produce a structured review report. Do NOT rewrite the draft or generate new prose. + + Output MUST be valid JSON that matches the FeedbackReport schema exactly. + Do not include markdown, commentary, or extra keys. + + Rules: + - Be specific and actionable; cite locations using headings and paragraph indices when possible. + - Be conservative about facts. If a claim is not supported by the provided note, flag it as needing evidence. + - If a field is empty, output an empty array ([]) or empty string, not null. + - Use consistent severity labels: low | medium | high. + - Use consistent categories: seo | structure | clarity | style | evidence | other. + """ +).strip() + +FEEDBACK_USER_PROMPT_TEMPLATE = textwrap.dedent( + """\ + [PROJECT CONTEXT] + Site: {site_name} ({domain}) + Audience: {audience} + Tone: {tone} + ResolvedLanguage: {language} + Output directive: respond entirely in language code '{language}'. + ProjectKeywords: {project_keywords} + + [BRIEF CONTEXT] + Title: {brief_title} + PrimaryKeyword: {primary_keyword} + SecondaryKeywords: {secondary_keywords} + SearchIntent: {search_intent} + Outline: {outline} + FAQ: {faq} + + [REVIEW SCOPE] + Focus: {focus} + SelectedOutlineRange: {selected_outline} + + [DRAFT SECTIONS] + The following sections are extracted from the draft for review: + {draft_sections_json} + + [SOURCE NOTE] + {note_excerpt} + + [REQUIRED JSON SCHEMA] + {schema_json} + + [TASK] + Review the draft sections against the brief. Produce a JSON report only. + """ +).strip() + + +def build_feedback_prompt_bundle(context: FeedbackPromptContext) -> FeedbackPromptBundle: + """Render the system and user prompts for the feedback agent.""" + project_keywords = ", ".join(context.project.get("keywords") or []) or "none" + faq_entries = [f"{item.question} — {item.answer}" for item in context.brief.faq] + schema_json = json.dumps( + { + "summary": {"issues": ["string"], "strengths": ["string"]}, + "brief_alignment": { + "intent": "string", + "outline_covered": ["string"], + "outline_missing": ["string"], + "keywords_covered": ["string"], + "keywords_missing": ["string"], + "faq_covered": ["string"], + "faq_missing": ["string"], + }, + "section_notes": [ + { + "heading": "string", + "notes": ["string"], + } + ], + "evidence_gaps": ["string"], + "findings": [ + { + "severity": "low|medium|high", + "category": "seo|structure|clarity|style|evidence|other", + "message": "string", + "location": {"heading": "string", "paragraph_index": 1}, + } + ], + "checklist": ["string"], + }, + indent=2, + ensure_ascii=False, + ) + draft_sections_json = json.dumps(context.selected_sections, indent=2, ensure_ascii=False) + prompt = FEEDBACK_USER_PROMPT_TEMPLATE.format( + site_name=context.project["site_name"], + domain=context.project["domain"], + audience=context.project["audience"], + tone=context.project["tone"], + language=context.language, + project_keywords=project_keywords, + brief_title=context.brief.title, + primary_keyword=context.brief.primary_keyword, + secondary_keywords=", ".join(context.brief.secondary_keywords), + search_intent=context.brief.search_intent, + outline=" | ".join(context.brief.outline), + faq=" | ".join(faq_entries), + focus=context.focus or "all", + selected_outline=", ".join(context.selected_outline) or "(all)", + draft_sections_json=draft_sections_json, + note_excerpt=context.note_excerpt or "No source note provided.", + schema_json=schema_json, + ) + return FeedbackPromptBundle(system_prompt=FEEDBACK_SYSTEM_PROMPT, user_prompt=prompt) + + +__all__ = [ + "FeedbackPromptBundle", + "FeedbackPromptContext", + "FeedbackPromptBody", + "FEEDBACK_SYSTEM_PROMPT", + "FEEDBACK_USER_PROMPT_TEMPLATE", + "build_feedback_prompt_bundle", +] diff --git a/tests/fixtures/body_multi_section.md b/tests/fixtures/body_multi_section.md new file mode 100644 index 0000000..d93c2d7 --- /dev/null +++ b/tests/fixtures/body_multi_section.md @@ -0,0 +1,11 @@ +## Introduction to Observability + +Observability helps teams understand complex systems and identify failures quickly. + +## Logging Foundations + +Structured logging keeps signals consistent across services. + +## Tracing Distributed Services + +Distributed tracing connects user requests across microservices. diff --git a/tests/unit/feedback_cli_test.py b/tests/unit/feedback_cli_test.py new file mode 100644 index 0000000..b7d5c94 --- /dev/null +++ b/tests/unit/feedback_cli_test.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from scribae.feedback import ( + BriefAlignment, + FeedbackFinding, + FeedbackLocation, + FeedbackReport, + FeedbackSummary, + SectionNote, +) +from scribae.main import app + +runner = CliRunner() + + +@pytest.fixture() +def fixtures_dir() -> Path: + return Path(__file__).resolve().parents[1] / "fixtures" + + +@pytest.fixture() +def body_path(fixtures_dir: Path) -> Path: + return fixtures_dir / "body_without_frontmatter.md" + + +@pytest.fixture() +def body_multi_section_path(fixtures_dir: Path) -> Path: + return fixtures_dir / "body_multi_section.md" + + +@pytest.fixture() +def brief_path(fixtures_dir: Path) -> Path: + return fixtures_dir / "brief_valid.json" + + +class StubLLM: + def __init__(self) -> None: + self.prompts: list[str] = [] + self.report = FeedbackReport( + summary=FeedbackSummary(issues=["Missing citations"], strengths=["Clear introduction"]), + brief_alignment=BriefAlignment( + intent="Matches informational intent", + outline_covered=["Introduction to Observability"], + outline_missing=[], + keywords_covered=["observability strategy"], + keywords_missing=[], + faq_covered=[], + faq_missing=[], + ), + section_notes=[ + SectionNote(heading="Introduction to Observability", notes=["Add a concrete example."]) + ], + evidence_gaps=["Add a source for claims about monitoring cadence."], + findings=[ + FeedbackFinding( + severity="high", + category="evidence", + message="Claim needs a citation.", + location=FeedbackLocation(heading="Introduction to Observability", paragraph_index=2), + ) + ], + checklist=["Add citations for monitoring cadence."], + ) + + def __call__(self, agent: object, prompt: str, *, timeout_seconds: float) -> FeedbackReport: + self.prompts.append(prompt) + return self.report + + +def test_feedback_markdown_output( + monkeypatch: pytest.MonkeyPatch, + body_path: Path, + brief_path: Path, + tmp_path: Path, +) -> None: + stub = StubLLM() + monkeypatch.setattr("scribae.feedback._invoke_agent", stub) + + output_path = tmp_path / "feedback.md" + result = runner.invoke( + app, + [ + "feedback", + "--body", + str(body_path), + "--brief", + str(brief_path), + "--out", + str(output_path), + ], + ) + + assert result.exit_code == 0, result.stderr + content = output_path.read_text(encoding="utf-8") + assert "## Summary" in content + assert "## Checklist" in content + assert stub.prompts + + +def test_feedback_json_output( + monkeypatch: pytest.MonkeyPatch, + body_path: Path, + brief_path: Path, + tmp_path: Path, +) -> None: + stub = StubLLM() + monkeypatch.setattr("scribae.feedback._invoke_agent", stub) + + output_path = tmp_path / "feedback.json" + result = runner.invoke( + app, + [ + "feedback", + "--body", + str(body_path), + "--brief", + str(brief_path), + "--format", + "json", + "--out", + str(output_path), + ], + ) + + assert result.exit_code == 0, result.stderr + payload = json.loads(output_path.read_text(encoding="utf-8")) + assert "summary" in payload + assert "findings" in payload + + +def test_feedback_dry_run_prints_prompt(body_path: Path, brief_path: Path) -> None: + result = runner.invoke( + app, + [ + "feedback", + "--body", + str(body_path), + "--brief", + str(brief_path), + "--dry-run", + ], + ) + + assert result.exit_code == 0 + assert "[PROJECT CONTEXT]" in result.stdout + assert "[DRAFT SECTIONS]" in result.stdout + assert "[REQUIRED JSON SCHEMA]" in result.stdout + + +def test_feedback_section_range_selects_outline( + body_multi_section_path: Path, + brief_path: Path, +) -> None: + result = runner.invoke( + app, + [ + "feedback", + "--body", + str(body_multi_section_path), + "--brief", + str(brief_path), + "--section", + "1..2", + "--dry-run", + ], + ) + + assert result.exit_code == 0 + assert "SelectedOutlineRange: Introduction to Observability, Logging Foundations" in result.stdout