File size: 8,334 Bytes
2c310c1
 
 
 
 
 
 
 
 
464ae11
 
2c310c1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
464ae11
 
 
 
 
 
2c310c1
 
464ae11
 
 
 
2c310c1
 
 
 
 
 
 
 
 
464ae11
 
 
 
 
 
 
2c310c1
 
464ae11
 
2c310c1
 
464ae11
 
 
 
 
2c310c1
 
 
 
 
 
 
 
 
 
 
 
 
464ae11
2c310c1
 
 
 
 
 
464ae11
 
 
 
2c310c1
 
 
 
464ae11
2c310c1
 
 
464ae11
 
2c310c1
464ae11
 
 
 
 
 
 
 
 
 
 
2c310c1
 
464ae11
 
 
 
 
 
 
 
 
 
2c310c1
464ae11
2c310c1
464ae11
 
 
2c310c1
464ae11
 
 
 
 
 
2c310c1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
"""Chat transcripts, one file per thread, in the per-student dataset storage.

Layout, alongside the profile that's already there:

    students/{student_id}/chats/index.json          β€” the thread list
    students/{student_id}/chats/{thread_id}.json    β€” one conversation

Every write is a git commit in the dataset repo (`server/storage.py`), so the
budget is **one write per completed turn** β€” never per token. The index is only
rewritten when the sidebar's ordering would actually change, which for a normal
conversation means once, when it starts.

This is also the shape the planned "My Story" feature needs: a durable,
timestamped record of what a student asked and what they were pointed toward.
"""
from __future__ import annotations

import logging
import re
import secrets
from datetime import datetime, timezone

from .. import storage

log = logging.getLogger("foresight.agent")

MAX_THREADS_LISTED = 50


def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def new_thread_id() -> str:
    return "t_" + secrets.token_hex(6)


def _valid(thread_id: str) -> bool:
    """Thread ids become path segments β€” never let one traverse."""
    return bool(re.fullmatch(r"t_[0-9a-f]{12}", thread_id or ""))


def _path(student_id: str, name: str) -> str:
    return storage.student_path(student_id, f"chats/{name}")


def title_from(question: str) -> str:
    """A thread title from the first question. Cheap on purpose β€” a second model
    call to name a conversation isn't worth the latency or the tokens."""
    words = (question or "").strip().split()
    title = " ".join(words[:7])
    if len(words) > 7:
        title += "…"
    return title[:80] or "New conversation"


# --- read -------------------------------------------------------------------
def load(student_id: str, thread_id: str) -> dict | None:
    """The transcript, or None if there isn't one to show.

    A deleted thread reads back as its tombstone (see `delete`), which is not a
    conversation β€” returning it would 200 an empty transcript to the client and
    let a new turn resurrect the thread the student threw away.
    """
    if not _valid(thread_id):
        return None
    data = storage.read_json(_path(student_id, f"{thread_id}.json"))
    if not data or data.get("deleted_at"):
        return None
    return data


def index(student_id: str) -> list[dict]:
    data = storage.read_json(_path(student_id, "index.json")) or {}
    threads = data.get("threads") or []
    threads.sort(key=lambda t: t.get("updated_at") or "", reverse=True)
    return threads[:MAX_THREADS_LISTED]


def most_recent_id(student_id: str) -> str | None:
    """The conversation to fall back to when a session doesn't know where it was
    β€” a fresh sign-in, or the same student on a second device."""
    listed = index(student_id)
    return listed[0].get("thread_id") if listed else None


# --- write ------------------------------------------------------------------
def append_turn(student_id: str, thread_id: str | None, question: str,
                answer: dict, partial: bool = False) -> dict:
    """Persist one exchange and return the saved thread.

    `answer` is the runner's final event: text, sources, suggestion, tools.

    `partial` marks an answer the student walked out on β€” they reloaded or closed
    the tab while it was still streaming, so what we have is however far it got.
    It's saved rather than dropped: the exchange was on their screen, and finding
    it missing when they come back is worse than finding it cut short.
    """
    created = False
    if not thread_id or not _valid(thread_id):
        thread_id, created = new_thread_id(), True

    thread = load(student_id, thread_id) if not created else None
    if thread is None:
        thread = {"thread_id": thread_id, "created_at": _now(),
                  "title": title_from(question), "messages": []}
        created = True

    stamp = _now()
    thread["messages"].append({"role": "user", "text": question, "at": stamp})
    reply = {
        "role": "assistant",
        "text": answer.get("text") or "",
        "at": stamp,
        "sources": answer.get("sources") or [],
        "suggestion": answer.get("suggestion"),
        "tools": answer.get("tools") or [],
    }
    if partial:
        reply["partial"] = True
    thread["messages"].append(reply)
    thread["updated_at"] = stamp

    storage.write_json(_path(student_id, f"{thread_id}.json"), thread,
                       message=f"chat: turn in {thread_id}")
    _touch_index(student_id, thread)
    return thread


def _touch_index(student_id: str, thread: dict) -> None:
    """Keep the index ordered by last activity, for as few commits as possible.

    An index write is a second commit per turn, so we only pay for one when the
    ordering would actually change: a brand-new thread, or a turn in a thread
    that isn't already at the top. Continuing the conversation you're already in
    β€” the common case by far β€” still costs one commit per turn, which is what the
    write budget here has always been.

    The consequence is that an entry's `updated_at` is not "last message"; it's
    "last time this thread moved to the top", which is exactly what ordering
    needs and nothing more. Don't render it as a last-activity time. `created_at`
    is carried alongside it precisely so the sidebar has an honest stamp to show
    ("Started …"), and `load()` always has the real per-message timestamps.
    """
    data = storage.read_json(_path(student_id, "index.json")) or {"threads": []}
    threads = data.get("threads") or []
    if threads and threads[0].get("thread_id") == thread["thread_id"]:
        return                      # already first; a rewrite would change nothing

    # `preview` is the opening question and `answer` the first reply β€” both are
    # the *first* of their kind, never the latest, so an entry stays true however
    # long the conversation runs and however rarely this function rewrites it.
    # The title is the question too, so a sidebar showing both lines wants the
    # answer on the second one; `preview` stays for entries written before that.
    preview = answer = ""
    for m in thread.get("messages", []):
        if not preview and m.get("role") == "user":
            preview = (m.get("text") or "")[:120]
        elif not answer and m.get("role") == "assistant":
            answer = " ".join((m.get("text") or "").split())[:140]
        if preview and answer:
            break
    rest = [t for t in threads if t.get("thread_id") != thread["thread_id"]]
    rest.insert(0, {"thread_id": thread["thread_id"], "title": thread["title"],
                    "created_at": thread.get("created_at"),
                    "updated_at": thread.get("updated_at"),
                    "preview": preview, "answer": answer})
    data["threads"] = rest[:MAX_THREADS_LISTED]
    try:
        storage.write_json(_path(student_id, "index.json"), data,
                           message=f"chat: index {thread['thread_id']}")
    except Exception:
        # The transcript is already saved; a missing index entry is recoverable
        # and shouldn't fail the student's turn.
        log.exception("chat: failed to update the thread index")


def delete(student_id: str, thread_id: str) -> bool:
    """Remove a thread from the index and blank its transcript.

    `storage` has no delete β€” it's an upload-only wrapper over the dataset repo β€”
    so the transcript is overwritten with a tombstone rather than removed. The
    repo's history keeps the old content either way; this is "gone from the app",
    not "erased from git".
    """
    if not _valid(thread_id):
        return False
    data = storage.read_json(_path(student_id, "index.json")) or {"threads": []}
    before = len(data.get("threads", []))
    data["threads"] = [t for t in data.get("threads", []) if t.get("thread_id") != thread_id]
    storage.write_json(_path(student_id, "index.json"), data,
                       message=f"chat: delete {thread_id}")
    storage.write_json(_path(student_id, f"{thread_id}.json"),
                       {"thread_id": thread_id, "deleted_at": _now(), "messages": []},
                       message=f"chat: tombstone {thread_id}")
    return len(data["threads"]) < before