workspace

workspace

Projects, habits and notes for the native Succès workspace.

The phase-two entities use the same local-first operation log as tasks. They stay on the Mac, keep tombstones for later replication, and materialize data that phase one deliberately preserved inside archived Life OS snapshots.

Classes

SuccesWorkspaceStore

SuccesWorkspaceStore(db_path: str | Path | None = None)

Bases: SuccesStore

Task store extended with the remaining phase-two workspace entities.

Source code in src/diapason/succes/workspace.py

def __init__(self, db_path: str | Path | None = None) -> None:
    super().__init__(db_path)
    with self._connect() as conn:
        conn.executescript(_WORKSPACE_SCHEMA)
        self._ensure_note_columns(conn)
        self._ensure_habit_columns(conn)
        conn.commit()
    self.materialize_archived_snapshots()

Methods:

create_task_edge

create_task_edge(
    project_id: str,
    from_task_id: str,
    to_task_id: str,
    *,
    op_id: str | None = None,
) -> dict[str, Any]

Relier deux tâches : « from débloque to ».

Le graphe doit rester sans boucle. Avec un cycle, « que puis-je faire maintenant ? » n'a plus de réponse — chaque tâche attend l'autre — et c'est précisément la question que la forme réseau existe pour poser. On refuse donc l'arête qui fermerait une boucle, en le disant.

Source code in src/diapason/succes/workspace.py

def create_task_edge(
    self,
    project_id: str,
    from_task_id: str,
    to_task_id: str,
    *,
    op_id: str | None = None,
) -> dict[str, Any]:
    """Relier deux tâches : « from débloque to ».

    Le graphe doit rester sans boucle. Avec un cycle, « que puis-je faire
    maintenant ? » n'a plus de réponse — chaque tâche attend l'autre — et
    c'est précisément la question que la forme réseau existe pour poser.
    On refuse donc l'arête qui fermerait une boucle, en le disant.
    """
    de, vers = str(from_task_id or "").strip(), str(to_task_id or "").strip()
    if not de or not vers:
        raise SuccesError("Une arête relie deux tâches : les deux sont requises.")
    if de == vers:
        raise SuccesError("Une tâche ne peut pas se débloquer elle-même.")
    request = {
        "action": "create_edge",
        "projectId": project_id,
        "fromTaskId": de,
        "toTaskId": vers,
    }
    ts = now_ms()
    with self._transaction() as conn:
        if self._load_project(conn, project_id) is None:
            raise SuccesNotFound("Ce projet n'existe pas ou a été supprimé.")
        self._assert_task_in_project(conn, de, project_id, role="amont")
        self._assert_task_in_project(conn, vers, project_id, role="aval")
        # Une boucle se formerait si « de » est déjà atteignable depuis
        # « vers » en suivant les arêtes existantes.
        atteints = {vers}
        frontiere = [vers]
        while frontiere:
            courant = frontiere.pop()
            for row in conn.execute(
                "SELECT to_task_id FROM succes_task_edges "
                "WHERE project_id=? AND from_task_id=?",
                (project_id, courant),
            ).fetchall():
                suivant = row["to_task_id"]
                if suivant == de:
                    raise SuccesError(
                        "Cette arête fermerait une boucle : chaque tâche "
                        "attendrait l'autre, et rien ne serait jamais "
                        "faisable."
                    )
                if suivant not in atteints:
                    atteints.add(suivant)
                    frontiere.append(suivant)
        existante = conn.execute(
            "SELECT updated_at_ms FROM succes_task_edges "
            "WHERE from_task_id=? AND to_task_id=?",
            (de, vers),
        ).fetchone()
        if existante is not None:
            # Rien n'a changé : ne pas enregistrer d'op. Un double-clic
            # gonflait le journal que le pair rejoue et rafraîchissait
            # updatedAtMs comme si l'arête venait de naître.
            return {
                "projectId": project_id,
                "fromTaskId": de,
                "toTaskId": vers,
                "updatedAtMs": int(existante["updated_at_ms"]),
            }
        conn.execute(
            "INSERT INTO succes_task_edges "
            "(project_id, from_task_id, to_task_id, updated_at_ms) "
            "VALUES (?,?,?,?)",
            (project_id, de, vers, ts),
        )
        edge = {
            "projectId": project_id,
            "fromTaskId": de,
            "toTaskId": vers,
            "updatedAtMs": ts,
        }
        self._record_op(
            conn,
            entity="task_edges",
            entity_id=f"{de}->{vers}",
            kind="upsert",
            payload=edge,
            request=request,
            timestamp_ms=ts,
            op_id=op_id,
        )
    return edge

list_project_tasks

list_project_tasks(project_id: str) -> list[dict[str, Any]]

Les tâches vivantes d'un projet, sous-tâches comprises.

list_tasks n'a jamais filtré par projet : l'outil vocal aurait résolu « budget » parmi les quatre-vingt-cinq tâches de toutes les listes, et relié deux tâches de projets différents.

Source code in src/diapason/succes/workspace.py

def list_project_tasks(self, project_id: str) -> list[dict[str, Any]]:
    """Les tâches vivantes d'un projet, sous-tâches comprises.

    `list_tasks` n'a jamais filtré par projet : l'outil vocal aurait
    résolu « budget » parmi les quatre-vingt-cinq tâches de toutes les
    listes, et relié deux tâches de projets différents.
    """
    self.get_project(project_id)
    with self._connect() as conn:
        ids = [
            row["id"]
            for row in conn.execute(
                "SELECT id FROM succes_tasks WHERE project_id=? "
                "AND deleted_at_ms IS NULL ORDER BY order_index, updated_at_ms",
                (project_id,),
            ).fetchall()
        ]
        return [task for tid in ids if (task := self._load_task(conn, tid))]

branches_de

branches_de(
    project_id: str, task_id: str
) -> dict[str, Any]

Les branches d'une tâche : amont, aval, la chaîne entière, ce que la terminer ouvre. Les champs sont ceux du fil (camelCase anglais).

Source code in src/diapason/succes/workspace.py

def branches_de(self, project_id: str, task_id: str) -> dict[str, Any]:
    """Les branches d'une tâche : amont, aval, la chaîne entière, ce que
    la terminer ouvre. Les champs sont ceux du fil (camelCase anglais)."""
    reseau = self._reseau(project_id)
    tid = str(task_id or "").strip()
    if tid not in reseau.par_id:
        raise SuccesNotFound(
            "Cette tâche n'existe pas dans ce projet ou a été supprimée."
        )
    amont = reseau_module.voisines_ordonnees(reseau, tid, "amont")
    aval = reseau_module.voisines_ordonnees(reseau, tid, "aval")
    return {
        "task": self._resume_tache(reseau, tid),
        "status": reseau_module.statut_de(reseau, tid),
        "upstream": [self._resume_tache(reseau, i) for i in amont],
        "downstream": [self._resume_tache(reseau, i) for i in aval],
        "upstreamAll": [
            {**self._resume_tache(reseau, i), "depth": p}
            for i, p in reseau_module.chaine(reseau, tid, "amont")
        ],
        "downstreamAll": [
            {**self._resume_tache(reseau, i), "depth": p}
            for i, p in reseau_module.chaine(reseau, tid, "aval")
        ],
        "missing": [
            self._resume_tache(reseau, i)
            for i in amont
            if not bool(reseau.par_id[i].get("done"))
        ],
        "unlocks": [
            self._resume_tache(reseau, i)
            for i in reseau_module.ce_que_debloque(reseau, tid)
        ],
        "impact": reseau_module.impact(reseau, tid),
    }

prochaines_actions

prochaines_actions(project_id: str) -> list[dict[str, Any]]

Les faisables maintenant, celles qui libèrent le plus d'abord.

Source code in src/diapason/succes/workspace.py

def prochaines_actions(self, project_id: str) -> list[dict[str, Any]]:
    """Les faisables maintenant, celles qui libèrent le plus d'abord."""
    reseau = self._reseau(project_id)
    return [
        {
            **self._resume_tache(reseau, tid),
            "impact": reseau_module.impact(reseau, tid),
            "unlocks": [
                self._resume_tache(reseau, i)
                for i in reseau_module.ce_que_debloque(reseau, tid)
            ],
        }
        for tid in reseau_module.faisables(reseau)
    ]

reset_cycle

reset_cycle(
    project_id: str, *, op_id: str | None = None
) -> dict[str, Any]

Décoche toutes les tâches du projet : un tour recommence.

Explicite, jamais automatique : une régénération déclenchée par une simple lecture serait un GET qui écrit, et l'utilisateur verrait ses coches disparaître sans geste de sa part.

Source code in src/diapason/succes/workspace.py

def reset_cycle(
    self, project_id: str, *, op_id: str | None = None
) -> dict[str, Any]:
    """Décoche toutes les tâches du projet : un tour recommence.

    Explicite, jamais automatique : une régénération déclenchée par une
    simple lecture serait un GET qui écrit, et l'utilisateur verrait ses
    coches disparaître sans geste de sa part.
    """
    project = self.get_project(project_id)
    if project.get("structure") != "cycle":
        raise SuccesError("Seul un projet en cycle recommence un tour.")
    request = {"action": "reset_cycle", "projectId": project_id}
    ts = now_ms()
    rouvertes = 0
    with self._transaction() as conn:
        # Rejouer un tour ne doit pas en refaire un : sans ce garde, un
        # client qui renvoie son op après une coupure décoche une seconde
        # fois un travail entre-temps refait.
        if op_id:
            deja = conn.execute(
                "SELECT 1 FROM succes_operations WHERE op_id LIKE ? LIMIT 1",
                (f"{op_id}:%",),
            ).fetchone()
            if deja is not None:
                return {
                    "project": self._load_project(conn, project_id),
                    "reopened": 0,
                }
        ids = [
            row["id"]
            for row in conn.execute(
                "SELECT id FROM succes_tasks "
                "WHERE project_id=? AND deleted_at_ms IS NULL AND done=1",
                (project_id,),
            ).fetchall()
        ]
        for task_id in ids:
            conn.execute(
                "UPDATE succes_tasks SET done=0, completed_date='', "
                "updated_at_ms=? WHERE id=?",
                (ts, task_id),
            )
            # Les sous-tâches suivent. Sans cela, elles restaient cochées :
            # au premier geste sur l'une d'elles, le recalcul voyait tout
            # l'arbre fait et re-cochait la tâche — le tour « recommencé »
            # s'achevait tout seul, sans que rien n'ait été fait.
            conn.execute(
                "UPDATE succes_subtasks SET done=0, updated_at_ms=? "
                "WHERE task_id=? AND deleted_at_ms IS NULL",
                (ts, task_id),
            )
            task = self._load_task(conn, task_id)
            assert task is not None
            # Une opération PAR tâche : la synchronisation rejoue des
            # tâches, pas des gestes de projet.
            self._record_op(
                conn,
                entity="tasks",
                entity_id=task_id,
                kind="upsert",
                payload=task,
                request=request,
                timestamp_ms=ts,
                op_id=f"{op_id}:{task_id}" if op_id else None,
            )
            rouvertes += 1
    return {"project": self.get_project(project_id), "reopened": rouvertes}

reorder_projects

reorder_projects(ids: list[str]) -> list[dict[str, Any]]

Fixe l'ordre manuel des projets : order_index = rang dans ids. Seuls les projets dont le rang change sont réécrits et journalisés, pour que le maillage voie le nouvel ordre sans inonder la relève.

Source code in src/diapason/succes/workspace.py

def reorder_projects(self, ids: list[str]) -> list[dict[str, Any]]:
    """Fixe l'ordre manuel des projets : order_index = rang dans `ids`.
    Seuls les projets dont le rang change sont réécrits et journalisés,
    pour que le maillage voie le nouvel ordre sans inonder la relève."""
    propres: list[str] = []
    for project_id in ids:
        pid = str(project_id or "").strip()
        if pid and pid not in propres:
            propres.append(pid)
    timestamp = now_ms()
    changes: list[dict[str, Any]] = []
    with self._transaction() as conn:
        rang = 0
        for project_id in propres:
            row = conn.execute(
                "SELECT order_index FROM succes_projects"
                " WHERE id=? AND deleted_at_ms IS NULL",
                (project_id,),
            ).fetchone()
            if row is None:
                continue
            index = rang
            rang += 1
            if int(row["order_index"]) == index:
                continue
            conn.execute(
                "UPDATE succes_projects SET order_index=?,updated_at_ms=?"
                " WHERE id=?",
                (index, timestamp, project_id),
            )
            project = self._load_project(conn, project_id)
            if project is not None:
                self._record_op(
                    conn,
                    entity="projects",
                    entity_id=project_id,
                    kind="upsert",
                    payload=project,
                    request={"action": "reorder_project", "order": index},
                    timestamp_ms=timestamp,
                )
                changes.append(project)
    return changes

list_habit_logs

list_habit_logs(
    *,
    from_date: str,
    to_date: str,
    habit_id: str | None = None,
) -> dict[str, bool]

Return done=true logs keyed as habitId_YYYY-MM-DD for a date span.

Source code in src/diapason/succes/workspace.py

def list_habit_logs(
    self,
    *,
    from_date: str,
    to_date: str,
    habit_id: str | None = None,
) -> dict[str, bool]:
    """Return done=true logs keyed as ``habitId_YYYY-MM-DD`` for a date span."""
    start = _validate_iso_date(from_date, "La date de début")
    end = _validate_iso_date(to_date, "La date de fin")
    if not start or not end:
        raise SuccesError("Indiquez une plage de dates complète.")
    if end < start:
        raise SuccesError("La date de fin doit suivre la date de début.")
    # Bound the window so a bad client cannot pull the entire history at once.
    if (date.fromisoformat(end) - date.fromisoformat(start)).days > 400:
        raise SuccesError("La plage de suivi ne peut pas dépasser 400 jours.")
    query = """SELECT habit_id, log_date FROM succes_habit_logs
               WHERE done=1 AND log_date>=? AND log_date<=?"""
    params: list[Any] = [start, end]
    if habit_id:
        query += " AND habit_id=?"
        params.append(habit_id)
    with self._connect() as conn:
        if habit_id and self._load_habit(conn, habit_id) is None:
            raise SuccesNotFound("Cette habitude n'existe pas ou a été supprimée.")
        rows = conn.execute(query, params).fetchall()
    return {f"{row['habit_id']}_{row['log_date']}": True for row in rows}

reorder_notes

reorder_notes(ids: list[str]) -> list[dict[str, Any]]

Fixe l'ordre manuel des notes citées : order_index = leur rang dans ids. Seules les notes DONT le rang change sont réécrites et journalisées — un glisser parmi dix ne doit pas inonder le journal de relève (§5). Une note absente garde son rang ; un id inconnu est ignoré en silence (la liste vient du réseau, elle n'est pas de confiance).

Source code in src/diapason/succes/workspace.py

def reorder_notes(self, ids: list[str]) -> list[dict[str, Any]]:
    """Fixe l'ordre manuel des notes citées : order_index = leur rang dans
    `ids`. Seules les notes DONT le rang change sont réécrites et
    journalisées — un glisser parmi dix ne doit pas inonder le journal de
    relève (§5). Une note absente garde son rang ; un id inconnu est ignoré
    en silence (la liste vient du réseau, elle n'est pas de confiance)."""
    propres: list[str] = []
    for note_id in ids:
        nid = str(note_id or "").strip()
        if nid and nid not in propres:
            propres.append(nid)
    timestamp = now_ms()
    changees: list[dict[str, Any]] = []
    with self._transaction() as conn:
        # Un id inconnu ne consomme PAS de rang : il ne pousserait pas les
        # notes réelles d'un cran. Le rang ne compte que les existantes.
        rang = 0
        for note_id in propres:
            row = conn.execute(
                "SELECT order_index FROM succes_notes"
                " WHERE id=? AND deleted_at_ms IS NULL",
                (note_id,),
            ).fetchone()
            if row is None:
                continue
            index = rang
            rang += 1
            if int(row["order_index"]) == index:
                continue
            conn.execute(
                "UPDATE succes_notes SET order_index=?,updated_at_ms=? WHERE id=?",
                (index, timestamp, note_id),
            )
            note = self._load_note(conn, note_id)
            if note is not None:
                self._record_op(
                    conn,
                    entity="notes",
                    entity_id=note_id,
                    kind="upsert",
                    payload=note,
                    request={"action": "reorder_note", "order": index},
                    timestamp_ms=timestamp,
                )
                changees.append(note)
    return changees

list_note_categories

list_note_categories() -> list[str]

Les catégories vivantes, dans l'ordre choisi par l'utilisateur.

Une catégorie EXISTE tant qu'une note vivante la porte — il n'y a pas d'état séparé à entretenir, donc pas d'orphelines. La table ne garde que l'ORDRE ; ses lignes mortes sont élaguées au passage, et une catégorie apparue depuis (note importée, synchronisée) se range à la fin, par ordre alphabétique.

Source code in src/diapason/succes/workspace.py

def list_note_categories(self) -> list[str]:
    """Les catégories vivantes, dans l'ordre choisi par l'utilisateur.

    Une catégorie EXISTE tant qu'une note vivante la porte — il n'y a pas
    d'état séparé à entretenir, donc pas d'orphelines. La table ne garde
    que l'ORDRE ; ses lignes mortes sont élaguées au passage, et une
    catégorie apparue depuis (note importée, synchronisée) se range à la
    fin, par ordre alphabétique.
    """
    with self._transaction() as conn:
        vivantes = {
            str(row["category"])
            for row in conn.execute(
                """SELECT DISTINCT category FROM succes_notes
                   WHERE deleted_at_ms IS NULL AND category != ''"""
            )
        }
        ordonnees = [
            str(row["name"])
            for row in conn.execute(
                "SELECT name FROM succes_note_categories ORDER BY order_index"
            )
        ]
        for morte in [n for n in ordonnees if n not in vivantes]:
            conn.execute(
                "DELETE FROM succes_note_categories WHERE name=?", (morte,)
            )
        gardees = [n for n in ordonnees if n in vivantes]
        return gardees + sorted(vivantes - set(gardees))

order_note_categories

order_note_categories(names: list[str]) -> list[str]

Mémorise l'ordre des sections — celui du glisser de Carlito.

Source code in src/diapason/succes/workspace.py

def order_note_categories(self, names: list[str]) -> list[str]:
    """Mémorise l'ordre des sections — celui du glisser de Carlito."""
    propres: list[str] = []
    for name in names:
        nom = _clean_text(name, field="La catégorie", maximum=60)
        if nom and nom not in propres:
            propres.append(nom)
    timestamp = now_ms()
    with self._transaction() as conn:
        for index, nom in enumerate(propres):
            conn.execute(
                """INSERT INTO succes_note_categories
                   (name,order_index,updated_at_ms)
                   VALUES (?,?,?)
                   ON CONFLICT(name) DO UPDATE
                   SET order_index=excluded.order_index,
                       updated_at_ms=excluded.updated_at_ms""",
                (nom, index, timestamp),
            )
    return self.list_note_categories()

rename_note_category

rename_note_category(ancien: str, nouveau: str) -> int

Renomme (ou dissout, si nouveau est vide) une catégorie entière.

Chaque note touchée est journalisée une à une : la synchronisation ne connaît que des notes, pas des catégories.

Source code in src/diapason/succes/workspace.py

def rename_note_category(self, ancien: str, nouveau: str) -> int:
    """Renomme (ou dissout, si `nouveau` est vide) une catégorie entière.

    Chaque note touchée est journalisée une à une : la synchronisation ne
    connaît que des notes, pas des catégories.
    """
    ancien_nom = _clean_text(ancien, field="La catégorie", maximum=60)
    nouveau_nom = _clean_text(nouveau or "", field="La catégorie", maximum=60)
    if not ancien_nom:
        raise SuccesError("La catégorie à renommer est obligatoire.")
    timestamp = now_ms()
    with self._transaction() as conn:
        notes = [
            row["id"]
            for row in conn.execute(
                """SELECT id FROM succes_notes
                   WHERE deleted_at_ms IS NULL AND category=?""",
                (ancien_nom,),
            )
        ]
        for note_id in notes:
            conn.execute(
                """UPDATE succes_notes SET category=?,updated_at_ms=?
                   WHERE id=?""",
                (nouveau_nom, timestamp, note_id),
            )
            note = self._load_note(conn, note_id)
            if note is not None:
                self._record_op(
                    conn,
                    entity="notes",
                    entity_id=note_id,
                    kind="upsert",
                    payload=note,
                    request={
                        "action": "rename_note_category",
                        "from": ancien_nom,
                        "to": nouveau_nom,
                    },
                    timestamp_ms=timestamp,
                )
        # L'ordre suit le nom ; une dissolution retire la ligne.
        if nouveau_nom:
            conn.execute(
                """UPDATE OR REPLACE succes_note_categories SET name=?
                   WHERE name=?""",
                (nouveau_nom, ancien_nom),
            )
        else:
            conn.execute(
                "DELETE FROM succes_note_categories WHERE name=?", (ancien_nom,)
            )
    return len(notes)

Functions: