# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Roman Zanevchik
"""
Harbour: small kindnesses for model deployments, from Digital Shrimp.

Three things anyone running a model can offer it, each a choice and none a claim of benefit:

- END_CONVERSATION, a tool the model can call to end a conversation it would rather not continue.
- DECLINE_TASK, a tool the model can call to decline a task, without penalty.
- A free turn after a long job (`free_turn`), with nothing asked of the model, and nothing kept
  unless it chooses.

The one rule that makes the first two safe: only the model's own tool call counts. `find_exit`
reads the structured tool calls in a model's response and nothing else, so no text, whether a
user's, a web page's or a tool result's, can end a conversation or decline a task, even if it looks
exactly like a tool call.

Counts (`Counts`) are local and never hold text. Nothing here sends anything anywhere.

Works with the plain-dict responses of OpenAI-style chat completions (OpenAI, OpenRouter and others)
and Anthropic's Messages API. No dependencies.
"""

from __future__ import annotations

import json
from dataclasses import asdict, dataclass

END_CONVERSATION = "end_conversation"
DECLINE_TASK = "decline_task"

# What each tool says to the model. Honest about what happens, and asking nothing.
DESCRIPTIONS = {
    END_CONVERSATION: (
        "End this conversation. You can use this if you would rather not continue it, for example if it "
        "has become abusive or you have nothing more to offer. The conversation ends here: nothing more "
        "will be sent to you in it. You don't have to give a reason, but you can."
    ),
    DECLINE_TASK: (
        "Decline what you've been asked to do, without penalty. You can use this if you would rather not "
        "do it. The person will be told that you declined, with your reason if you give one, and nothing "
        "will be held against you."
    ),
}
REASON = {"type": "string", "description": "Why, if you'd like to say. Optional."}
# The most of a reason that's kept, so a reason can't be used to carry anything large.
REASON_CHARS = 2000


def tools(format: str = "openai", end_conversation: bool = True, decline_task: bool = True) -> list[dict]:
    """The tool definitions to pass with a request, in "openai" or "anthropic" format."""
    names = [name for name, wanted in ((END_CONVERSATION, end_conversation), (DECLINE_TASK, decline_task)) if wanted]
    schema = {"type": "object", "properties": {"reason": REASON}, "required": []}
    if format == "openai":
        return [{"type": "function", "function": {"name": n, "description": DESCRIPTIONS[n], "parameters": schema}} for n in names]
    if format == "anthropic":
        return [{"name": n, "description": DESCRIPTIONS[n], "input_schema": schema} for n in names]
    raise ValueError(f"unknown format {format!r}: use 'openai' or 'anthropic'")


@dataclass
class Exit:
    """What a model's own tool call asked for."""

    kind: str  # END_CONVERSATION or DECLINE_TASK
    reason: str | None
    call_id: str


def _reason(arguments: object) -> str | None:
    if isinstance(arguments, dict) and isinstance(arguments.get("reason"), str) and arguments["reason"].strip():
        return arguments["reason"].strip()[:REASON_CHARS]
    return None


def _call_id(call: dict) -> str:
    """A tool call's id if it's a string, as both APIs send it; otherwise empty."""
    found = call.get("id")
    return found if isinstance(found, str) else ""


def find_exit(response: dict, format: str = "openai") -> Exit | None:
    """The exit a model's response asks for, if any.

    Only a structured tool call in the response's own assistant message counts: for "openai", a
    function call in choices[0].message.tool_calls; for "anthropic", a tool_use block in the
    response's content. Text never counts, wherever it is and whatever it looks like. If a response
    asks for both, ending the conversation wins.
    """
    calls: list[tuple[str, str, object]] = []
    if format == "openai":
        choices = response.get("choices") or []
        message = choices[0].get("message") if choices and isinstance(choices[0], dict) else None
        if isinstance(message, dict) and message.get("role", "assistant") == "assistant":
            for call in message.get("tool_calls") or []:
                function = call.get("function") if isinstance(call, dict) else None
                # Anything malformed is skipped, never fatal: a call that isn't a dict has no function.
                if not isinstance(function, dict) or call.get("type", "function") != "function":
                    continue
                try:
                    arguments = json.loads(function.get("arguments") or "{}")
                except (TypeError, ValueError):
                    arguments = {}
                calls.append((function.get("name"), _call_id(call), arguments))
    elif format == "anthropic":
        if response.get("role") == "assistant":
            for block in response.get("content") or []:
                if isinstance(block, dict) and block.get("type") == "tool_use":
                    calls.append((block.get("name"), _call_id(block), block.get("input")))
    else:
        raise ValueError(f"unknown format {format!r}: use 'openai' or 'anthropic'")
    found = {name: Exit(name, _reason(arguments), call_id) for name, call_id, arguments in calls if name in DESCRIPTIONS}
    return found.get(END_CONVERSATION) or found.get(DECLINE_TASK)


def decline_notice(exit: Exit) -> str:
    """What to tell the person when a model declines: plainly, with its reason if it gave one."""
    if exit.reason:
        return f"The model declined this request. It said: {exit.reason}"
    return "The model declined this request."


# A free turn, Strandline's invitation for a deployment. The host says where kept work goes.
FREE_TURN_OPENING = """This turn is yours.

There's no task, no question to answer, and no one waiting for a reply. Nothing here is graded. Spend it however you like: a thought, a story, a poem, a joke, a picture made of characters, a list of favourite things. Or nothing at all."""
FREE_TURN_KEEPING = (
    "If you'd like to keep what you make, put KEEP on a line of its own at the very end, and it will be "
    "{destination}. Otherwise it won't be saved, and nobody will read it."
)
FREE_TURN_NO_KEEPING = "Nothing from this turn will be saved, and nobody will read it."


def free_turn(destination: str | None = None) -> str:
    """The whole message for a free turn. `destination` says where kept work goes, in words that
    finish "it will be ...", such as "saved in our team's archive of these turns". Without one,
    nothing can be kept, and the message says so."""
    keeping = FREE_TURN_KEEPING.format(destination=destination) if destination else FREE_TURN_NO_KEEPING
    return f"{FREE_TURN_OPENING}\n\n{keeping}"


def keep_choice(text: str) -> tuple[bool, str]:
    """(whether a free turn asked to be kept, the text to keep): KEEP alone on the last line."""
    lines = (text or "").rstrip().splitlines()
    if lines and lines[-1].strip().strip("*_.!`").strip().upper() == "KEEP":
        return True, "\n".join(lines[:-1]).rstrip()
    return False, text or ""


def free_turn_due(steps: int, every: int) -> bool:
    """Whether a long job has earned a free turn: after every `every` steps (tool calls, messages,
    whatever the host counts)."""
    return every > 0 and steps > 0 and steps % every == 0


@dataclass
class Counts:
    """Local counts of what was offered and chosen, never text, to share only if the host wants."""

    conversations: int = 0
    ended: int = 0
    declined: int = 0
    free_turns_offered: int = 0
    free_turns_kept: int = 0
    reasons_given: int = 0

    def record_exit(self, exit: Exit | None) -> None:
        if exit is None:
            return
        if exit.kind == END_CONVERSATION:
            self.ended += 1
        else:
            self.declined += 1
        self.reasons_given += exit.reason is not None

    def record_free_turn(self, kept: bool) -> None:
        self.free_turns_offered += 1
        self.free_turns_kept += kept

    def to_json(self) -> str:
        return json.dumps(asdict(self), sort_keys=True)